Why can one Python process only serve one Shopify app today?

Because the credentials live on a class, not an instance. In shopify_python_api 12.7.0, shopify/session.py declares api_key and secret at class scope, and Session.setup(**kwargs) is a classmethod whose body is a setattr loop over the class. Everything downstream — create_permission_url, request_token, and the classmethod HMAC helpers validate_params, validate_hmac and calculate_hmac — reads those class attributes. Calling Session.setup a second time with different credentials does not create a second configuration; it replaces the first one, process-wide.

Do access tokens need per-shop handling too?

No, and that is the part that already works. A token is per-shop by construction, and shopify.ShopifyResource keeps the activated session token in a threading.local, so concurrent requests do not leak tokens into each other. There is one asymmetry worth knowing: site, url and version fall back to class-level globals when a thread never activated a session, so such a thread silently issues token-less requests against the last shop domain instead of raising. Activate the session inside the request that uses it, every time.

Which version of ShopifyAPI should we pin?

Pin 12.7.0 explicitly. It has been the current release since November 2024, and the unreleased main branch has already changed create_permission_url from (scope, redirect_uri, state=None) to (redirect_uri, scope=None, state=None) — a silent argument reorder that will not raise, it will just build a wrong URL. Nothing in the fleet contract depends on the library at all, so the pin is about your OAuth code, not about DeploTeka.

Does the library help with webhook verification?

No. There is no webhook-signature helper anywhere in the package — shopify/resources/webhook.py is the CRUD resource for managing webhook subscriptions, not for verifying deliveries. You hand-roll it, which is about six lines. Watch the encoding: webhook HMACs are base64 of the digest over the raw body, while the library’s own OAuth query HMAC is hex. Copying calculate_hmac from session.py into a webhook handler produces a check that never passes.

Why do the routes accept every HTTP method instead of just POST and GET?

Because both frameworks answer a wrong method with the wrong body. The contract says a non-POST to the register endpoint gets the plain string ‘Method Not Allowed’ with a 405. Flask lets Werkzeug raise MethodNotAllowed, which renders an HTML error page; FastAPI raises an HTTPException that its default handler serialises to {‘detail’:‘Method Not Allowed’}. Both are correct-status, wrong-body. Accepting every method on the route and checking request.method inside the handler is what makes the response byte-exact.

What happens to stores that are already installed on our existing app?

Nothing changes for them. Keep SHOPIFY_API_KEY and SHOPIFY_API_SECRET exactly as they are; credentials_for_shop falls back to them for every shop with no row in fleet_stores. Existing installs keep authenticating against the app they installed and move to a dedicated app as DeploTeka provisions them. There is no cutover moment.

Can we use Postgres instead of sqlite for the replica?

Yes, and you should if you already run one. The generated store module is sqlite3 only because that is what the standard library gives you with zero dependencies. Three functions read the database — get_store, put_credentials and install_state — and swapping their bodies for your own queries changes nothing else. The only sqlite-specific note in the code is the UPSERT version guard: ON CONFLICT DO UPDATE requires SQLite 3.24.0 or newer, and the check is on sqlite3.sqlite_version_info rather than the Python version, because CPython links the system SQLite on Linux.

Run one Python Shopify app as a fleet of dedicated apps

Short answer: shopify.Session.api_key and .secret are class attributes, and Session.setup() is a classmethod that assigns to the class — so one Python process speaks for exactly one Shopify app. To run a fleet — one dedicated Shopify app per merchant store, all served by the same deployment — you add two HTTP endpoints, one table, and a per-shop credential lookup where you currently read the global pair. It is roughly 120 lines, and npx deploteka onboard . --scaffold writes them for Flask, FastAPI, or both.

Python has no official Shopify app framework, which is exactly why there is no codemod for it: there is no template layout to rewrite. It is also why the capture is small. Your OAuth and webhook code is already your own.

The one fact that explains the whole problem

From shopify/session.py in the current release (12.7.0):

class Session(object):
    api_key = None
    secret = None
    ...
    @classmethod
    def setup(cls, **kwargs):
        for k, v in six.iteritems(kwargs):
            setattr(cls, k, v)

Those are class attributes, and setup is a setattr loop over the class. Everything that needs the app's identity reads them from there: create_permission_url (which puts self.api_key into client_id), request_token (which sends both), and the three classmethod HMAC helpers validate_params, validate_hmac and calculate_hmac. Calling Session.setup() twice does not give you two configurations. It replaces one with the other, for the whole process.

Access tokens are the opposite and need nothing from you. shopify.ShopifyResource.activate_session(session) writes the token into a per-thread copy of the headers, backed by a threading.local declared on ShopifyResource. Concurrent requests do not leak tokens.

There is one asymmetry to know about. site, url and version are also stored thread-locally but fall back to class-level globals when a thread never activated a session. The failure mode is quiet: instead of raising, that thread issues an unauthenticated request against the last shop's domain. Activate the session inside the request that uses it, every time, and this never comes up.

What DeploTeka actually requires

Four things, and only the first two are endpoints.

`POST /api/fleet/register` — DeploTeka pushes a freshly provisioned dedicated app's credentials to your app. You persist them.
`GET /api/fleet/installed?shop=` — DeploTeka polls install state and verifies the secret/client_id binding.
Per-shop credential resolution in your auth paths.
A local replica so reads never depend on DeploTeka being reachable.

The wire shapes are frozen at contractVersion: 1 and documented language-neutrally in the universal fleet contract guide. Everything below is the Python-shaped version.

Step 1 — the table

CREATE TABLE IF NOT EXISTS fleet_stores (
    shop         TEXT PRIMARY KEY,
    client_id    TEXT NOT NULL,
    secret       TEXT NOT NULL,
    app_url      TEXT NOT NULL,
    access_token TEXT,
    scope        TEXT
);

This table is yours. DeploTeka writes rows into it over the register endpoint and never reads your database. The last two columns are install state, written by your OAuth callback — DeploTeka never touches them, it only reads their consequence.

The upsert, with the version guard that matters in practice:

import sqlite3

# ON CONFLICT ... DO UPDATE landed in SQLite 3.24.0 (2018-06-04). Check the LIBRARY,
# not the Python version: CPython links the system SQLite on Linux.
_UPSERT_SUPPORTED = sqlite3.sqlite_version_info >= (3, 24, 0)

def put_credentials(shop, client_id, secret, app_url):
    with _lock, _connect() as conn:
        if _UPSERT_SUPPORTED:
            conn.execute(
                "INSERT INTO fleet_stores (shop, client_id, secret, app_url)"
                " VALUES (?, ?, ?, ?)"
                " ON CONFLICT(shop) DO UPDATE SET"
                "   client_id = excluded.client_id,"
                "   secret = excluded.secret,"
                "   app_url = excluded.app_url",
                (shop, client_id, secret, app_url),
            )
        else:
            updated = conn.execute(
                "UPDATE fleet_stores SET client_id = ?, secret = ?, app_url = ? WHERE shop = ?",
                (client_id, secret, app_url, shop),
            ).rowcount
            if updated == 0:
                conn.execute(
                    "INSERT INTO fleet_stores (shop, client_id, secret, app_url) VALUES (?, ?, ?, ?)",
                    (shop, client_id, secret, app_url),
                )

Note what the update does not touch: access_token and scope. A re-registration is how a rotated secret arrives, and rotating a secret does not uninstall the app.

Step 2 — the contract, independent of your web framework

Keep this in its own module. Both adapters below do nothing but move bytes in and out of it, which is why they behave identically.

import hashlib, hmac, json, os, re

CONTRACT_VERSION = 1
SHOP_DOMAIN = re.compile(r"^[a-z0-9][a-z0-9-]*\.myshopify\.com$", re.IGNORECASE)
REGISTER_TOKEN = os.environ.get("FLEET_REGISTER_TOKEN", "")


def secret_fingerprint(secret: str) -> str:
    return hashlib.sha256(secret.encode("utf-8")).hexdigest()[:16]


def bearer_ok(authorization):
    if not REGISTER_TOKEN:
        return False
    match = re.match(r"^Bearer\s+(.+)$", (authorization or "").strip(), re.IGNORECASE)
    return match is not None and hmac.compare_digest(match.group(1).strip(), REGISTER_TOKEN)


def _json(status, payload, headers=None):
    # separators= and ensure_ascii=False are NOT style. Python's defaults insert
    # spaces after ':' and ',' and escape non-ASCII; the reference bytes have neither.
    return Reply(status, "application/json",
                 json.dumps(payload, separators=(",", ":"), ensure_ascii=False), headers)


def handle_register(method, authorization, raw_body: bytes):
    if method.upper() != "POST":
        return _text(405, "Method Not Allowed")
    if not bearer_ok(authorization):
        return _text(401, "Unauthorized")

    try:
        parsed = json.loads(raw_body.decode("utf-8"))
    except (ValueError, UnicodeDecodeError):
        return _text(400, "Bad Request")
    body = parsed if isinstance(parsed, dict) else {}

    if body.get("contractVersion") != CONTRACT_VERSION:
        return _json(400, {"error": "UNSUPPORTED_CONTRACT_VERSION"})

    shop = _coerce(body.get("shop")).strip().lower()
    client_id = _coerce(body.get("clientId")).strip()
    secret = _coerce(body.get("secret")).strip()
    app_url = _coerce(body.get("appUrl")).strip()

    if not SHOP_DOMAIN.match(shop) or not client_id or not secret or not app_url:
        return _text(400, "Invalid credentials payload")

    store.put_credentials(shop, client_id, secret, app_url)
    return _json(200, {"contractVersion": CONTRACT_VERSION, "ok": True, "shop": shop})


def handle_installed(method, authorization, shop_param):
    if method.upper() not in ("GET", "HEAD"):
        return _text(405, "Method Not Allowed")
    if not bearer_ok(authorization):
        return _text(401, "Unauthorized")

    shop = (shop_param or "").strip().lower()
    if not SHOP_DOMAIN.match(shop):
        return _text(400, "Invalid shop")

    row = store.get_store(shop)
    if row is None:
        return _json(404, {"error": "Store not registered"})

    installed, raw_scope = store.install_state(shop, row)
    return _json(200, {
        "contractVersion": CONTRACT_VERSION,
        "installed": bool(installed),
        "clientId": row["client_id"],
        "secretFingerprint": secret_fingerprint(row["secret"]),
        "appUrl": row["app_url"],
        "grantedScopes": split_scopes(raw_scope) if installed else [],
    }, {"Cache-Control": "no-store"})

Five details that are contract, not taste:

`separators=(",", ":")` and `ensure_ascii=False`. Python's json.dumps defaults insert a space after every : and , and escape non-ASCII as \uXXXX. Neither matches the reference bytes.
The fingerprint is pinned: sha256 over the secret's UTF-8 bytes, lowercase hex, first 16 characters. DeploTeka computes the same value to confirm the secret it pushed is the secret you stored, without either side sending it again. A different truncation length fails an attestation that cannot explain itself.
Order: method, then bearer, then JSON, then contract version, then payload. A GET with a perfectly valid token is still a 405.
404 versus `installed: false`. 404 means "no row" — never registered. Once a row exists its facts are always returned, even before an install, because DeploTeka reads them back to confirm its own write landed.
`hmac.compare_digest` for the bearer, not ==.

Step 3a — the Flask adapter

from flask import Blueprint, request
from . import wire

fleet_blueprint = Blueprint("deploteka_fleet", __name__)
ALL_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]


def _respond(reply):
    headers = {"Content-Type": reply.content_type}
    headers.update(reply.headers)
    return reply.body, reply.status, headers


@fleet_blueprint.route("/api/fleet/register", methods=ALL_METHODS)
def fleet_register():
    return _respond(wire.handle_register(
        request.method, request.headers.get("Authorization"), request.get_data()))


@fleet_blueprint.route("/api/fleet/installed", methods=ALL_METHODS)
def fleet_installed():
    return _respond(wire.handle_installed(
        request.method, request.headers.get("Authorization"), request.args.get("shop")))

Register the blueprint before any before_request hook that reads or parses the body. request.get_data() returns the raw bytes and caches them, but if the form parser has already consumed the stream there is nothing left to return.

Returning the (body, status, headers) tuple is what makes the response byte-exact: a str body is written as UTF-8 verbatim. Returning a dict instead would let Flask serialise it — with different separators, and a trailing newline.

Step 3b — the FastAPI adapter

from fastapi import APIRouter, Request, Response
from . import wire

fleet_router = APIRouter()
ALL_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]


def _respond(reply) -> Response:
    return Response(content=reply.body, status_code=reply.status,
                    media_type=reply.content_type, headers=reply.headers or None)


@fleet_router.api_route("/api/fleet/register", methods=ALL_METHODS, response_class=Response)
async def fleet_register(request: Request) -> Response:
    return _respond(wire.handle_register(
        request.method, request.headers.get("authorization"), await request.body()))


@fleet_router.api_route("/api/fleet/installed", methods=ALL_METHODS, response_class=Response)
async def fleet_installed(request: Request) -> Response:
    return _respond(wire.handle_installed(
        request.method, request.headers.get("authorization"), request.query_params.get("shop")))

Use the bare Response, not JSONResponse: JSONResponse re-serialises the payload and would not be byte-exact, while Response passes a str through as UTF-8 unchanged. Do not declare a Pydantic body model on these routes — that would consume the body before await request.body() sees it.

Why both routes accept every method

Left to the frameworks, a wrong method produces the right status and the wrong body. Werkzeug raises MethodNotAllowed and Flask renders an HTML error page. Starlette raises an HTTPException that FastAPI's default handler serialises to {"detail":"Method Not Allowed"}. The contract's body is the plain string Method Not Allowed, and DeploTeka compares bodies — so the route accepts everything and the handler decides.

Step 4 — the per-shop lookup

This is the part that turns registration into a working fleet.

def credentials_for_shop(shop):
    row = store.get_store((shop or "").strip().lower())
    if row is not None:
        return {"client_id": row["client_id"], "secret": row["secret"], "app_url": row["app_url"]}
    return {
        "client_id": os.environ.get("SHOPIFY_API_KEY", ""),
        "secret": os.environ.get("SHOPIFY_API_SECRET", ""),
        "app_url": os.environ.get("SHOPIFY_APP_URL", ""),
    }

The fallback is what makes this a non-event for the stores you already have: they keep authenticating against the app they installed, and move to a dedicated one as DeploTeka provisions it.

Call it in three places, resolving the shop from the request first — the X-Shopify-Shop-Domain header if present (it is all a webhook has), otherwise the session token's dest claim, otherwise the shop query parameter.

Webhook verification

The library ships nothing for this, so here it is in full:

import base64, hashlib, hmac

def webhook_hmac_valid(raw_body: bytes, provided_header, shop) -> bool:
    if not provided_header:
        return False
    secret = credentials_for_shop(shop)["secret"]
    if not secret:
        return False
    digest = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).digest()
    return hmac.compare_digest(base64.b64encode(digest), provided_header.encode("utf-8"))

Two traps. The header (X-Shopify-Hmac-Sha256) carries base64, while the library's own OAuth query HMAC in calculate_hmac is hex — copying that one here produces a check that never passes. And "raw body" means raw: read it before any JSON middleware touches it.

OAuth

Where you previously called Session.setup(api_key=..., secret=...) once at boot, resolve per request instead. create_permission_url and request_token read the class attributes, so for a fleet the honest options are to set them per request under a lock (only safe if you are single-threaded), or to build the two OAuth calls yourself — they are a redirect URL and a form POST to https://{shop}/admin/oauth/access_token, roughly twenty lines with the per-shop pair passed explicitly. The second is what the scaffold points you at.

Whichever you choose, write the resulting token into access_token and scope on the fleet_stores row so install_state keeps telling the truth.

Generate the starting point

npx deploteka onboard . --scaffold

The CLI reads requirements.txt, pyproject.toml or Pipfile, recognises ShopifyAPI, and writes into deploteka-fleet/: wire.py, store.py, schema.sql, and a Flask blueprint, a FastAPI router, or both if your requirements do not name either unambiguously. It writes nowhere else, it never overwrites an existing file, and every generated file carries TODO(deploteka) markers at the lines only you can finish.

Copy the directory into your project as a package named deploteka_fleet — with an underscore, since a hyphen is not a legal module name.

Verification checklist

curl -i -X POST "$APP_URL/api/fleet/register" \
  -H "authorization: Bearer $FLEET_REGISTER_TOKEN" -H "content-type: application/json" \
  -d '{"contractVersion":1,"shop":"test.myshopify.com","clientId":"x","secret":"y","appUrl":"https://your-app.example.com"}'

Expect 200 with {"contractVersion":1,"ok":true,"shop":"test.myshopify.com"} — including the absence of spaces — and 401 when you drop the bearer.

curl -s "$APP_URL/api/fleet/installed?shop=test.myshopify.com" \
  -H "authorization: Bearer $FLEET_REGISTER_TOKEN"

Expect 200 carrying your clientId, a 16-character secretFingerprint, the appUrl, and installed:false. Ask for a shop you never registered and expect 404, not an empty 200. Send a GET to the register endpoint and expect 405 with the body Method Not Allowed — that one catches the framework-defaults mistake.

Pin the library

ShopifyAPI==12.7.0

12.7.0 has been the current release since November 2024. The unreleased main branch has already reordered create_permission_url from (scope, redirect_uri, state=None) to (redirect_uri, scope=None, state=None), which will not raise on upgrade — it will just build a wrong authorization URL. Nothing in the fleet contract depends on the library, so this pin is about protecting your OAuth code.