Tenant Query

1"""Fetch all rows from `entities` for a given tenant.
2
3req.tenant is untrusted input — it MUST be validated before interpolation.
4The execution path here builds a raw SQL string (for cases where a driver's
5native parameter binding cannot be used — e.g., dynamic identifiers, shell
6pipelines). safe_query.build() is the injection defense.
7
8Authorization note: format validation (regex) confirms the value looks like a
9valid tenant slug. It does NOT prove the caller is authorized to read that
10tenant's data. Authorize the caller against req.tenant before calling this
11function.
12"""
13
14from safe_query import build, regex, ident, TENANT_SLUG
15
16
17def select_by_tenant(cursor, req) -> list[tuple]:
18    """Return all rows from `entities` where tenant_id matches req.tenant.
19
20    Uses cursor-based execution so the same shape works across psycopg2,
21    psycopg3, and pgx-style cursors (psycopg2's `cursor.execute` returns None,
22    so the caller must use `cursor.fetchall()` separately rather than chain).
23
24    Args:
25        cursor: A driver cursor that exposes `.execute(sql)` and `.fetchall()`.
26            For psycopg3 you can also pass `connection.cursor()` or use
27            `connection.execute(sql).fetchall()` directly inline.
28        req: An object with a `.tenant` attribute (untrusted string).
29
30    Returns:
31        A list of row tuples (or row dicts, depending on the cursor's
32        configured row factory).
33
34    Raises:
35        UnsafeSQLError: If req.tenant fails TENANT_SLUG validation.
36        ValueError: If req.tenant is missing or not a string.
37    """
38    tenant = getattr(req, "tenant", None)
39    if not isinstance(tenant, str):
40        raise ValueError(f"req.tenant must be a str, got {type(tenant).__name__}")
41
42    sql = build(
43        "SELECT * FROM {t} WHERE tenant_id = {tid}",
44        t=ident("entities"),
45        tid=regex(tenant, TENANT_SLUG, label="req.tenant"),
46    )
47
48    cursor.execute(sql)
49    return cursor.fetchall()