initial commit

Author: jcarbaugh

.github/workflows/tests.yml

@@ -0,0 +1,44 @@
+name: Tests
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_DB: test_fqq
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    env:
+      DATABASE_URL: postgres://postgres:postgres@localhost:5432/test_fqq
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Set up Python
+        run: uv python install
+
+      - name: Install dependencies
+        run: uv sync
+
+      - name: Run tests
+        run: uv run pytest

.gitignore

@@ -0,0 +1,14 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+.env
+
+.claude/

.python-version

@@ -0,0 +1 @@
+3.12

README.md

@@ -0,0 +1,133 @@
+# django-fqq
+
+Lightweight PostgreSQL schema-based multi-tenancy for Django.
+
+django-fqq automatically qualifies table names with the appropriate PostgreSQL schema, routing queries to tenant-specific schemas using a simple `ContextVar`-based approach. Shared tables (like `auth`) stay in `public`, while tenant tables are directed to the active schema.
+
+## How it works
+
+django-fqq provides a custom database backend that wraps Django's built-in PostgreSQL backend. It overrides `quote_name` to prepend the active schema to table names — so `"my_table"` becomes `"tenant_schema"."my_table"` automatically. No changes to your models or queries required.
+
+django-fqq does not modify queries if `set_schema` has not been called or has been set to a falsy value (e.g. None or "").
+
+## Installation
+
+```bash
+pip install django-fqq
+```
+
+## Configuration
+
+### 1. Set the database backend
+
+```python
+DATABASES = {
+    "default": {
+        "ENGINE": "django_fqq.backend",
+        "NAME": "your_db",
+        # ... other options
+    }
+}
+```
+
+### 2. Add the app
+
+```python
+INSTALLED_APPS = [
+    # ...
+    "django_fqq",
+]
+```
+
+### 3. Configure shared apps
+
+Apps listed in `FQQ_SHARED_APPS` will always use the `public` schema. Everything else uses the active tenant schema.
+
+```python
+FQQ_SHARED_APPS = ["auth", "contenttypes"]
+```
+
+### 4. Add the middleware (optional)
+
+```python
+MIDDLEWARE = [
+    "django_fqq.middleware.SchemaMiddleware",
+    # ...
+]
+```
+
+Subclass `SchemaMiddleware` and override `_get_schema(request)` to resolve the tenant schema from the request (e.g. from subdomain, header, or URL).
+
+## Usage
+
+### Setting the schema manually
+
+```python
+from django_fqq.schema import set_schema, clear_schema
+
+set_schema("tenant_abc")
+# All queries now target the "tenant_abc" schema
+# ...
+clear_schema()
+```
+
+### In a middleware or view
+
+```python
+from django_fqq.middleware import SchemaMiddleware
+
+class MySchemaMiddleware(SchemaMiddleware):
+    def _get_schema(self, request):
+        # Resolve tenant from subdomain
+        host = request.get_host().split(".")[0]
+        return host
+```
+
+### Using a context manager
+
+`query_schema` is a context manager that sets the schema for the duration of a block and automatically restores the previous state on exit. If a schema was already active, it is restored rather than cleared.
+
+```python
+from django_fqq.schema import query_schema
+
+with query_schema("tenant_abc"):
+    # All queries target "tenant_abc"
+    ...
+# Schema is restored to its previous value (or cleared if none was set)
+```
+
+It supports nesting — each level restores the schema that was active before it:
+
+```python
+from django_fqq.schema import set_schema, query_schema
+
+set_schema("tenant_a")
+
+with query_schema("tenant_b"):
+    # queries target "tenant_b"
+    with query_schema("tenant_c"):
+        # queries target "tenant_c"
+        ...
+    # back to "tenant_b"
+
+# back to "tenant_a"
+```
+
+The previous schema is also restored if an exception is raised inside the block.
+
+### Context-safe
+
+Schema state is stored in a `ContextVar`, so it's safe across concurrent async requests and threads.
+
+## Development
+
+Requires Python 3.12+ and Django 6.0+.
+
+```bash
+uv sync
+uv run pytest
+```
+
+## License
+
+MIT

django_fqq/__init__.py

@@ -0,0 +1,31 @@
+from django.apps import AppConfig
+
+
+class DjangoFqqConfig(AppConfig):
+    name = "django_fqq"
+    default_auto_field = "django.db.models.BigAutoField"
+
+    def ready(self):
+        from django.apps import apps
+        from django.conf import settings
+        from django.db import connections
+
+        from django_fqq.backend.operations import DatabaseOperations
+
+        shared_apps = set(getattr(settings, "FQQ_SHARED_APPS", []))
+
+        table_names = set()
+        shared_table_names = set()
+        for model in apps.get_models():
+            table = model._meta.db_table
+            table_names.add(table)
+            if model._meta.app_label in shared_apps:
+                shared_table_names.add(table)
+
+        DatabaseOperations._table_names = table_names
+        DatabaseOperations._shared_table_names = shared_table_names
+
+        for conn in connections.all():
+            if isinstance(conn.ops, DatabaseOperations):
+                conn.ops._table_names = table_names
+                conn.ops._shared_table_names = shared_table_names

django_fqq/apps.py

@@ -0,0 +1,9 @@
+from django.db.backends.postgresql.base import (
+    DatabaseWrapper as PostgresqlDatabaseWrapper,
+)
+
+from django_fqq.backend.operations import DatabaseOperations
+
+
+class DatabaseWrapper(PostgresqlDatabaseWrapper):
+    ops_class = DatabaseOperations

django_fqq/backend/__init__.py

@@ -0,0 +1,31 @@
+from django.db.backends.postgresql.operations import (
+    DatabaseOperations as PostgresqlDatabaseOperations,
+)
+
+from django_fqq.schema import get_schema
+
+
+class DatabaseOperations(PostgresqlDatabaseOperations):
+    _table_names: set[str] = set()
+    _shared_table_names: set[str] = set()
+
+    def quote_name(self, name):
+
+        schema = get_schema()
+        if not schema:
+            return super().quote_name(name)
+
+        if name.startswith('"'):
+            return name
+
+        if "." in name:
+            parts = name.split(".")
+            quote = super().quote_name
+            return ".".join(quote(part) for part in parts)
+
+        if name in self._table_names:
+            if name in self._shared_table_names:
+                schema = "public"
+            return super().quote_name(schema) + "." + super().quote_name(name)
+
+        return super().quote_name(name)

django_fqq/backend/base.py

@@ -0,0 +1,20 @@
+from django_fqq.schema import clear_schema, set_schema
+
+
+class SchemaMiddleware:
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        # TBD: determine schema from request (e.g. subdomain, header, etc.)
+        schema = self._get_schema(request)
+        if schema:
+            set_schema(schema)
+        try:
+            return self.get_response(request)
+        finally:
+            clear_schema()
+
+    def _get_schema(self, request):
+        """Override this method or replace with your own logic."""
+        return None