Skip to content

Resolution Strategies

Base

BaseTenantResolver

Python
BaseTenantResolver(store: TenantStore[Tenant])

Bases: ABC

Optional abstract base class for tenant resolution strategies.

Subclass this to build a custom resolution strategy::

Text Only
class CookieResolver(BaseTenantResolver):
    async def resolve(self, request: Request) -> Tenant:
        tenant_id = request.cookies.get("X-Tenant")
        if not tenant_id:
            raise TenantResolutionError("Cookie missing", strategy="cookie")
        return await self.store.get_by_identifier(tenant_id)

Alternatively, implement the TenantResolver protocol directly — duck-typing is sufficient (no inheritance required).

Parameters:

Name Type Description Default
store TenantStore[Tenant]

The tenant metadata store used to look up tenants.

 required
Source code in src/fastapi_tenancy/resolution/base.py
Python
48
49
def __init__(self, store: TenantStore[Tenant]) -> None:
    self.store = store

resolve abstractmethod async

Python
resolve(request: Request) -> Tenant

Resolve the current tenant from request.

Parameters:

Name Type Description Default
request Request

Incoming FastAPI / Starlette request.

 required

Returns:

Type Description
Tenant

The resolved :class:~fastapi_tenancy.core.types.Tenant.

Raises:

Type Description
TenantResolutionError

When the request does not carry enough information to identify a tenant.
TenantNotFoundError

When the identifier matches no known tenant.
Source code in src/fastapi_tenancy/resolution/base.py
Python
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@abstractmethod
async def resolve(self, request: Request) -> Tenant:
    """Resolve the current tenant from *request*.

    Args:
        request: Incoming FastAPI / Starlette request.

    Returns:
        The resolved :class:`~fastapi_tenancy.core.types.Tenant`.

    Raises:
        TenantResolutionError: When the request does not carry enough
            information to identify a tenant.
        TenantNotFoundError: When the identifier matches no known tenant.
    """

HeaderTenantResolver

Python
HeaderTenantResolver(
    store: TenantStore[Tenant],
    header_name: str = "X-Tenant-ID",
)

Bases: BaseTenantResolver

Resolve the current tenant from an HTTP request header.

Parameters:

Name Type Description Default
store TenantStore[Tenant]

Tenant metadata store.

 required
header_name str

Header to read (default: "X-Tenant-ID").

 'X-Tenant-ID'

Example::

Text Only
resolver = HeaderTenantResolver(store, header_name="X-Tenant-ID")

# Request: GET /api/users HTTP/1.1
#          X-Tenant-ID: acme-corp
tenant = await resolver.resolve(request)
# → Tenant(identifier="acme-corp", …)
Source code in src/fastapi_tenancy/resolution/header.py
Python
56
57
58
59
60
61
62
def __init__(
    self,
    store: TenantStore[Tenant],
    header_name: str = "X-Tenant-ID",
) -> None:
    super().__init__(store)
    self._header_name = header_name

resolve async

Python
resolve(request: Request) -> Tenant

Resolve the tenant from the X-Tenant-ID header (or configured name).

All failure modes (missing header, invalid identifier format, unknown tenant) raise :exc:TenantResolutionError with the same generic reason to prevent tenant enumeration by callers.

Parameters:

Name Type Description Default
request Request

Incoming HTTP request.

 required

Returns:

Name Type Description
Resolved Tenant

class:~fastapi_tenancy.core.types.Tenant.

Raises:

Type Description
TenantResolutionError

When the header is absent, fails identifier validation, or no matching tenant exists.
Source code in src/fastapi_tenancy/resolution/header.py
Python
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
async def resolve(self, request: Request) -> Tenant:
    """Resolve the tenant from the ``X-Tenant-ID`` header (or configured name).

    All failure modes (missing header, invalid identifier format, unknown
    tenant) raise :exc:`TenantResolutionError` with the same generic reason
    to prevent tenant enumeration by callers.

    Args:
        request: Incoming HTTP request.

    Returns:
        Resolved :class:`~fastapi_tenancy.core.types.Tenant`.

    Raises:
        TenantResolutionError: When the header is absent, fails identifier
            validation, or no matching tenant exists.
    """
    identifier = request.headers.get(self._header_name, "").strip()
    if not identifier:
        logger.debug("Header resolver: header %r missing or empty", self._header_name)
        raise TenantResolutionError(reason=_GENERIC_REASON, strategy="header")
    if not validate_tenant_identifier(identifier):
        logger.debug("Header resolver: invalid identifier format %r", identifier)
        raise TenantResolutionError(reason=_GENERIC_REASON, strategy="header")
    logger.debug("Header resolver: identifier=%r", identifier)
    try:
        return await self.store.get_by_identifier(identifier)
    except TenantNotFoundError:
        # Re-raise as TenantResolutionError with the same generic message
        # so the middleware returns 400 (not 404) for unknown tenants —
        # a 404 would confirm to callers that the identifier format is valid.
        raise TenantResolutionError(reason=_GENERIC_REASON, strategy="header")  # noqa: B904

Subdomain

SubdomainTenantResolver

Python
SubdomainTenantResolver(
    store: TenantStore[Tenant],
    domain_suffix: str = "",
    trust_x_forwarded: bool = True,
)

Bases: BaseTenantResolver

Resolve the current tenant from the leftmost subdomain.

Parameters:

Name Type Description Default
store TenantStore[Tenant]

Tenant metadata store.

 required
domain_suffix str

The base domain (e.g. ".example.com"). Used to strip the suffix before validation; if the host does not end with this suffix the resolver raises.

 ''
trust_x_forwarded bool

Whether to read X-Forwarded-Host before Host. Default True (assume trusted reverse proxy).

 True

Example::

Text Only
resolver = SubdomainTenantResolver(store, domain_suffix=".example.com")

# Request: Host: acme-corp.example.com
tenant = await resolver.resolve(request)
# → Tenant(identifier="acme-corp", …)
Source code in src/fastapi_tenancy/resolution/subdomain.py
Python
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    store: TenantStore[Tenant],
    domain_suffix: str = "",
    trust_x_forwarded: bool = True,
) -> None:
    super().__init__(store)
    # Normalise: always starts with "." unless empty.
    self._domain_suffix = (
        domain_suffix
        if not domain_suffix or domain_suffix.startswith(".")
        else f".{domain_suffix}"
    )
    self._trust_x_forwarded = trust_x_forwarded

resolve async

Python
resolve(request: Request) -> Tenant

Extract the tenant identifier from the request's hostname.

Parameters:

Name Type Description Default
request Request

Incoming HTTP request.

 required

Returns:

Name Type Description
Resolved Tenant

class:~fastapi_tenancy.core.types.Tenant.

Raises:

Type Description
TenantResolutionError

When the subdomain is absent, does not match the configured suffix, or fails validation.
TenantNotFoundError

When the identifier has no matching tenant.
Source code in src/fastapi_tenancy/resolution/subdomain.py
Python
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def resolve(self, request: Request) -> Tenant:
    """Extract the tenant identifier from the request's hostname.

    Args:
        request: Incoming HTTP request.

    Returns:
        Resolved :class:`~fastapi_tenancy.core.types.Tenant`.

    Raises:
        TenantResolutionError: When the subdomain is absent, does not
            match the configured suffix, or fails validation.
        TenantNotFoundError: When the identifier has no matching tenant.
    """
    host = ""
    if self._trust_x_forwarded:
        host = request.headers.get("x-forwarded-host", "")
    if not host:
        host = request.headers.get("host", "")
    if not host:
        raise TenantResolutionError(
            reason="Neither Host nor X-Forwarded-Host header is present",
            strategy="subdomain",
        )

    identifier = self._extract_identifier(host)
    logger.debug("Subdomain resolver: host=%r → identifier=%r", host, identifier)
    return await self.store.get_by_identifier(identifier)

Path

PathTenantResolver

Python
PathTenantResolver(
    store: TenantStore[Tenant],
    path_prefix: str = "/tenants",
)

Bases: BaseTenantResolver

Resolve the current tenant from the URL path prefix.

Parameters:

Name Type Description Default
store TenantStore[Tenant]

Tenant metadata store.

 required
path_prefix str

The fixed path prefix before the tenant identifier (default: "/tenants").

 '/tenants'

Example::

Text Only
resolver = PathTenantResolver(store, path_prefix="/tenants")

# Request: GET /tenants/acme-corp/orders
tenant = await resolver.resolve(request)
# → Tenant(identifier="acme-corp", …)
# request.state.tenant_path_remainder == "/orders"
Source code in src/fastapi_tenancy/resolution/path.py
Python
59
60
61
62
63
64
65
66
def __init__(
    self,
    store: TenantStore[Tenant],
    path_prefix: str = "/tenants",
) -> None:
    super().__init__(store)
    # Normalise: strip trailing slash, ensure leading slash.
    self._prefix = "/" + path_prefix.strip("/")

resolve async

Python
resolve(request: Request) -> Tenant

Extract the tenant identifier from the request path.

Parameters:

Name Type Description Default
request Request

Incoming HTTP request.

 required

Returns:

Name Type Description
Resolved Tenant

class:~fastapi_tenancy.core.types.Tenant.

Raises:

Type Description
TenantResolutionError

When the path does not match the expected format or the identifier fails validation.
TenantNotFoundError

When the identifier has no matching tenant.
Source code in src/fastapi_tenancy/resolution/path.py
Python
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def resolve(self, request: Request) -> Tenant:
    """Extract the tenant identifier from the request path.

    Args:
        request: Incoming HTTP request.

    Returns:
        Resolved :class:`~fastapi_tenancy.core.types.Tenant`.

    Raises:
        TenantResolutionError: When the path does not match the expected
            format or the identifier fails validation.
        TenantNotFoundError: When the identifier has no matching tenant.
    """
    path = request.url.path
    prefix_with_slash = self._prefix.rstrip("/") + "/"

    if not path.startswith(prefix_with_slash):
        raise TenantResolutionError(
            reason=(f"Path {path!r} does not start with expected prefix {prefix_with_slash!r}"),
            strategy="path",
        )

    remainder = path[len(prefix_with_slash) :]
    # The identifier is the first path segment after the prefix.
    identifier = remainder.split("/")[0]
    if not identifier:
        raise TenantResolutionError(
            reason=f"No tenant identifier found after prefix {self._prefix!r}",
            strategy="path",
        )
    if not validate_tenant_identifier(identifier):
        raise TenantResolutionError(
            reason=f"Path segment {identifier!r} is not a valid tenant identifier",
            strategy="path",
        )

    # Store the remainder (path after the tenant segment) on request state
    # for downstream handlers that need it.
    path_after_tenant = remainder[len(identifier) :]
    request.state.tenant_path_remainder = path_after_tenant or "/"

    logger.debug("Path resolver: path=%r → identifier=%r", path, identifier)
    return await self.store.get_by_identifier(identifier)

JWT

JWTTenantResolver

Python
JWTTenantResolver(
    store: TenantStore[Tenant],
    secret: str,
    algorithm: str = "HS256",
    tenant_claim: str = "tenant_id",
    audience: str | None = None,
)

Bases: BaseTenantResolver

Resolve the current tenant from a signed Bearer JWT.

Reads Authorization: Bearer <token> from the request, verifies the signature, and extracts the configured claim (default: tenant_id).

Parameters:

Name Type Description Default
store TenantStore[Tenant]

Tenant metadata store.

 required
secret str

JWT signing secret (HMAC) or public key (RSA). Required.

 required
algorithm str

Signing algorithm (default: "HS256").

 'HS256'
tenant_claim str

JWT payload claim holding the tenant identifier (default: "tenant_id").

 'tenant_id'
audience str | None

Expected aud claim value. When set, PyJWT verifies that the decoded token contains a matching audience claim and raises TenantResolutionError otherwise. Strongly recommended when the same JWT secret is shared across multiple services to prevent cross-service token replay attacks. Default: None (no audience check — a warning is emitted at resolver construction time).

 None

Raises:

Type Description
ImportError

When PyJWT is not installed.

Example::

Text Only
resolver = JWTTenantResolver(
    store,
    secret="my-super-secret-key-at-least-32-chars",
    tenant_claim="tenant_id",
    audience="my-api-service",
)

# Request: Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6Ikp...
tenant = await resolver.resolve(request)
Source code in src/fastapi_tenancy/resolution/jwt.py
Python
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __init__(
    self,
    store: TenantStore[Tenant],
    secret: str,
    algorithm: str = "HS256",
    tenant_claim: str = "tenant_id",
    audience: str | None = None,
) -> None:
    super().__init__(store)
    try:
        import jwt as _pyjwt  # noqa: PLC0415

        self._jwt = _pyjwt
    except ImportError as exc:
        raise ImportError(
            "JWT resolution requires 'PyJWT>=2.8'. "
            "Install it with: pip install 'fastapi-tenancy[jwt]'"
        ) from exc

    self._secret = secret
    self._algorithm = algorithm
    self._tenant_claim = tenant_claim
    self._audience = audience

    # Warn when no audience is configured so operators are
    # alerted to the cross-service replay risk during startup.
    if audience is None:
        logger.warning(
            "JWTTenantResolver: no 'audience' configured.  If multiple "
            "services share the same JWT secret, set audience= to prevent "
            "cross-service token replay attacks."
        )

resolve async

Python
resolve(request: Request) -> Tenant

Decode the Bearer JWT and resolve the tenant from the payload claim.

Parameters:

Name Type Description Default
request Request

Incoming HTTP request.

 required

Returns:

Name Type Description
Resolved Tenant

class:~fastapi_tenancy.core.types.Tenant.

Raises:

Type Description
TenantResolutionError

When the Authorization header is absent, malformed, or the JWT is invalid / expired.
TenantNotFoundError

When the extracted identifier has no matching tenant.
Source code in src/fastapi_tenancy/resolution/jwt.py
Python
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
async def resolve(self, request: Request) -> Tenant:
    """Decode the Bearer JWT and resolve the tenant from the payload claim.

    Args:
        request: Incoming HTTP request.

    Returns:
        Resolved :class:`~fastapi_tenancy.core.types.Tenant`.

    Raises:
        TenantResolutionError: When the ``Authorization`` header is absent,
            malformed, or the JWT is invalid / expired.
        TenantNotFoundError: When the extracted identifier has no
            matching tenant.
    """
    auth_header = request.headers.get("authorization", "")
    if not auth_header:
        raise TenantResolutionError(
            reason="Authorization header is missing",
            strategy="jwt",
        )
    if not auth_header.startswith(_BEARER_PREFIX):
        raise TenantResolutionError(
            reason="Authorization header does not use Bearer scheme",
            strategy="jwt",
        )

    token = auth_header[len(_BEARER_PREFIX) :].strip()
    if not token:
        raise TenantResolutionError(
            reason="Bearer token is empty",
            strategy="jwt",
        )

    payload = self._decode_token(token)

    identifier = payload.get(self._tenant_claim)
    if not identifier or not isinstance(identifier, str):
        raise TenantResolutionError(
            reason=f"JWT payload is missing claim {self._tenant_claim!r}",
            strategy="jwt",
            details={"claim": self._tenant_claim},
        )
    if not validate_tenant_identifier(identifier):
        raise TenantResolutionError(
            reason=f"JWT claim {self._tenant_claim!r} contains an invalid tenant identifier",
            strategy="jwt",
            details={"claim": self._tenant_claim},
        )

    logger.debug("JWT resolver: claim=%r → identifier=%r", self._tenant_claim, identifier)
    return await self.store.get_by_identifier(identifier)