diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml index 0b1d60cb..252b5d2c 100644 --- a/.github/workflows/pr.yml +++ b/.github/workflows/pr.yml @@ -217,6 +217,12 @@ jobs: # never in the generic module editor" rule is actually exercised — the # server-side half of that is a 409 guard, which is worth CI coverage. SM_MODULES_ENABLED: '["Auth","Users","Dashboard","Permissions","Settings","BackgroundTasks","FileStorage","FeatureFlags","AuditLog","Branding"]' + # Every test signs in from scratch, all from 127.0.0.1: the login page, + # the login POST and the landing page are anonymous public-route hits, + # ~120 a minute for the whole suite — right at the default + # ``rate_limit_public`` of 120/minute, so a fast runner 429s whichever + # test logs in last. The limiter itself is covered by unit tests. + SM_RATE_LIMIT_PUBLIC: 1000/minute E2E_BASE_URL: http://localhost:8000 steps: - uses: actions/checkout@v7 diff --git a/CHANGELOG.md b/CHANGELOG.md index f3ba8a52..f3f81d4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,19 @@ All notable changes to this project are documented in this file. The format is b ## [Unreleased] ### Added +- `from background_tasks import heartbeat` (#451): a long-running task calls + `heartbeat()` to refresh `heartbeat_at` (throttled to one write per 30 s; a + no-op outside a task). `@shared_task(..., stuck_after=)` sets a + per-task stuck threshold that the sweep honours ahead of + `SM_BG_TASKS_STUCK_AFTER_SECONDS`. No migration. +- `from background_tasks import enqueue_after_commit` (#452): publishes a task + only once the caller's transaction commits (nothing on rollback), and + returns the reserved task id so a row can name it up front. +- HTTP `Range` support on file downloads (#453): single ranges answer `206` + (`416` when unsatisfiable, multi-range falls back to `200`), honouring + `If-Range`. Storage backends gain `get_range(key, start, end)` (inclusive + end); a third-party backend without it keeps serving full `200`s. +- `smpy host gen-i18n` and a scaffold `make gen-i18n` target (#449). - **Host override layer for module copy** (#415). A host ships `host/locales/overrides/.json` (nested or flat dotted keys) and it is applied after every module, framework and host catalog, so the server-side @@ -166,6 +179,40 @@ All notable changes to this project are documented in this file. The format is b `tenants` resolver it selects among the user's own memberships. ### Fixed +- Timezone-aware timestamp columns read back aware on SQLite (#468). Columns + declared `sa_type=DateTime(timezone=True)` (the `AuditMixin`/`SoftDeleteMixin` + timestamps, audit log, users, permissions, tenant invitations) came back + naive, so the API sent them with no offset and browsers showed UTC as local + time. New `simple_module_db.AwareDateTime` replaces that column type; it + migrates as `sa.DateTime(timezone=True)`, so no schema change. +- Background-task Retry sends one request per execution while one is in + flight (#467); a burst of confirm clicks no longer queues duplicate retries. +- The background-tasks list shows its plain empty state when no executions + exist at all, instead of claiming some exist outside the search (#469). +- Deleting files another tab already deleted says so, instead of a success + toast reading "0 files deleted" (#470). +- `smpy`'s workspace-member hint prints `./modules/...` for a module inside + the host directory, so the suggested `smpy add` reads it as a path, not a + PyPI name (#471). +- Scaffolded compose apps require a URL-safe `POSTGRES_PASSWORD`: compose puts + it unencoded into `SM_DATABASE_URL`, so `make docker-up` now refuses other + characters with a clear message (#472). +- `S3Backend.get` streams objects in 64 KiB reads instead of buffering the + whole object in memory (#453). It is now awaited like the filesystem + backend, which also fixes S3 thumbnails and the streamed public-file path + (they raised `TypeError`), and a missing object answers `404`, not `500`. +- Admin task retries publish to the broker only after the new + `TaskExecution` rows commit (#452); a rolled-back request no longer leaves a + queued task with no execution row. +- `POST /api/users/auth/token` persists the upgraded password hash, as the + browser login already did (#457), so bcrypt hashes imported from another + system migrate to Argon2 on API/mobile sign-in too. +- Scaffolded apps regenerate `keys.generated.ts` in development (#449): when + `packages/i18n/src` is absent the host writes into the installed + `@simple-module-py/i18n` package, merging the keys it already ships. +- Scaffolded `make kill` stops only this app's ports (`API_PORT`, and + `VITE_PORT` read from `SM_VITE_DEV_URL`) instead of `pkill`-ing every + uvicorn and vite on the machine, vitest runs included (#461). - `render_item` renders sqlmodel>=0.0.48's `UTCDateTime` as `sa.DateTime(timezone=True)` and adds `import sqlmodel` for any other sqlmodel type, so an initial autogenerate no longer emits a migration that diff --git a/docs/framework/i18n.md b/docs/framework/i18n.md index 90636b65..5a2177db 100644 --- a/docs/framework/i18n.md +++ b/docs/framework/i18n.md @@ -283,6 +283,8 @@ The detection logic lives in `scripts/lib/untranslated-strings.mjs` behind its o `packages/i18n/src/keys.generated.ts` is generated over every **installed** module, not just the ones the running host activated, and `make ci-js-typecheck` runs `tsc -p` for each `modules/*/tsconfig.json` and `packages/*/tsconfig.json` in the workspace. (A module shipping `.tsx` with no `tsconfig.json` fails the target outright, rather than being silently skipped — the same gap `SM017` reports.) +**Scaffolded apps.** There is no `packages/i18n` checkout in a scaffolded app: `@simple-module-py/i18n` is an npm dependency. When `packages/i18n/src` is absent, development boot (and `make gen-i18n` / `smpy host gen-i18n`) writes `keys.generated.ts` and `generated-resources.ts` into the nearest `node_modules/@simple-module-py/i18n/src` found by walking up from the project root (so a hoisted npm-workspace install is found; a flat host also checks `client_app/node_modules`). The shared `ui.*` and framework keys come from catalogs inside `simple_module_hosting`, and any keys already in the published file are kept (the published catalog covers modules this app does not install, so they are not treated as removals). `npm install` restores the published file; `make install` regenerates afterwards. With neither location present (a wheel-installed production app) boot skips silently; `make gen-i18n` fails loudly. + That distinction is load-bearing: with a filtered union, translating an inactive auth provider's pages (say `keycloak`, when this host runs `users`) would break its build on the next regeneration while the app itself ran fine. The runtime registry stays filtered — an inactive module's strings are typed, never served. `t()` needs literal keys to typecheck; a key assembled at runtime won't resolve. diff --git a/docs/guide/installation.md b/docs/guide/installation.md index e32702be..f926ba92 100644 --- a/docs/guide/installation.md +++ b/docs/guide/installation.md @@ -89,7 +89,7 @@ POSTGRES_PASSWORD= SM_DATABASE_URL=postgresql+asyncpg://postgres:@localhost:5432/my_app ``` -The database name is the snake_case app name. `POSTGRES_PASSWORD` is shared with the compose `postgres` service below. +The database name is the snake_case app name. `POSTGRES_PASSWORD` is shared with the compose `postgres` service below. Compose interpolates it unencoded into the container's `SM_DATABASE_URL`, so it must be URL-safe (letters, digits, `-`, `_`, `.`, `~`); `make docker-up` refuses to start otherwise. The generated value always is. Every scaffold ships a `docker-compose.yml` that matches your `--db` choice (a migration history is dialect-frozen at autogenerate time, so containers must run the same database the migrations were generated against). With `--db postgres` it brings up `postgres` (db named after your app, user `postgres`, password from `POSTGRES_PASSWORD` in `.env`) on `127.0.0.1:5432` and the `app` service; the default SQLite scaffold runs the `app` container alone with the database on a named volume. Including the `background_tasks` module adds `redis` on `127.0.0.1:6379` plus `worker`/`beat` services that reuse the app image. With `--db postgres`, the container also doubles as a local dev database: @@ -165,7 +165,7 @@ Pass `--loosen` to rewrite every constraint to `>=` instead. Run `uv tool update-shell` (or restart your shell) so the tool's bin dir is on PATH. **Port already in use.** -Free `:8000` and `:5050` before the next `make dev`. On Linux/macOS: `lsof -ti:8000,5050 | xargs kill -9`. +Run `make kill` to free the app's API and Vite ports (`:8000` and the port in `SM_VITE_DEV_URL`, default `:5050`) before the next `make dev`. It only touches those ports, never other apps' servers. To run two apps side by side, give each its own `API_PORT` (`make dev API_PORT=8010`) and `SM_VITE_DEV_URL`. **Alembic complains about a revision mismatch.** The DB is ahead of or behind the migration files. For a dev DB: `rm app.db && make migrate`. For Postgres: `docker compose down -v && docker compose up -d postgres && make migrate`. diff --git a/docs/modules/background_tasks.md b/docs/modules/background_tasks.md index 9af5d253..c32f5129 100644 --- a/docs/modules/background_tasks.md +++ b/docs/modules/background_tasks.md @@ -32,6 +32,63 @@ send_receipt.delay(order_id=42) Every dispatch / start / completion / failure / retry / revoke is recorded in the `background_tasks_task_execution` table by Celery signal handlers, so you can see history even after the Celery result backend has expired the result. +## Enqueueing from inside a transaction + +`send_receipt.delay(...)` publishes *now*. If the request that called it is still writing the row the task will read, two things go wrong: a fast worker (or eager mode, which runs the task inside the publish) starts before the row is visible, and if the commit then fails the task is queued against a row that never existed. Publish after commit instead: + +```python +from background_tasks import enqueue_after_commit + + +async def place(self, data: OrderCreate) -> OrderOut: + order = Order(**data.model_dump()) + self.db.add(order) + await self.db.flush() + enqueue_after_commit(self.db, send_receipt, order.id) + return OrderOut.model_validate(order) +``` + +`enqueue_after_commit(db, task, *args, **kwargs)` sends `task` once `db`'s transaction commits and drops it if the transaction rolls back or the commit fails. It returns the Celery task id the message will carry, reserved up front, so you can store it on your own row in the same transaction. + +- `task` is a task (`send_receipt`), a signature for routing options (`send_receipt.s().set(queue="mail", countdown=30)`), or a registered task name (`"orders.send_receipt"`, sent through `celery.current_app`). +- In a request (`get_db`), the publish runs from the session's `on_commit` hook — after the commit, before the response leaves. +- Any other session — a task body's sync session, a CLI script's own `async with session.begin()` — hangs it off SQLAlchemy's `after_commit` event instead, and drops it when the outermost transaction ends without committing. Publishing then blocks the committing thread for the length of one kombu publish. +- The task runs as the tenant bound when you called `enqueue_after_commit`, not whatever is bound at commit time. +- A publish that fails after the commit is logged, not raised: the transaction is already durable. + +The admin retry buttons follow the same rule — the new `pending` row commits first, then the task is sent (GH #452). An orphaned `pending` row that was never sent is visible and retryable; a task with no row is neither. + +## Long-running tasks: heartbeats and stuck thresholds + +The beat sweep flips a `running` row to `stuck` once its `heartbeat_at` is older than the stuck threshold (`stuck_after_seconds`, default 300), which makes it retry-eligible. The signal handlers only stamp the heartbeat when a task starts and finishes, so a task that runs longer than the threshold must say it is still alive, or it can be retried — and run twice — while the first run is still working. + +Call `heartbeat()` from the task body's loop: + +```python +from background_tasks import heartbeat +from celery import shared_task + + +@shared_task(name="reports.rebuild") +def rebuild(report_id: int) -> None: + for chunk in chunks(report_id): + process(chunk) + heartbeat() +``` + +It refreshes `heartbeat_at` on the current execution's row (found through Celery's `current_task`), writing at most once every 30 seconds per task however often you call it, so calling it on every iteration is fine; `heartbeat(force=True)` skips the throttle. Outside a running task it does nothing and returns `False`, so shared helpers can call it unconditionally. It only touches a row that is still `running` — a heartbeat never revives a row the sweep already declared stuck. + +A task whose steps are legitimately longer than the global threshold can declare its own, in seconds: + +```python +@shared_task(name="exports.full_dump", stuck_after=4 * 3600) +def full_dump() -> None: ... +``` + +Celery copies extra decorator options onto the task class; the sweep reads `stuck_after` off the task registry of the worker running it (which autodiscovers every module's tasks), and falls back to `SM_BG_TASKS_STUCK_AFTER_SECONDS` / the default for tasks without one. No column or migration is involved, so the threshold always matches the deployed code. A shorter `stuck_after` works too, for tasks that should never take long. Keep any threshold well above 30 seconds, or the heartbeat throttle alone can let a live task look stale. + +Not provided: an automatic heartbeat thread. A task that blocks in one long call (a single huge query, an external process) cannot heartbeat from inside it — give it a `stuck_after` that covers the call instead. + ## Scheduling periodic work (beat) A module schedules recurring work the same way it registers tasks — by shipping a `tasks.py` — and additionally exporting a module-level `BEAT_SCHEDULE` dict. `build_celery` merges every installed module's `BEAT_SCHEDULE` into the beat schedule at boot (identically in the web and worker processes), so `make beat` runs them alongside the built-ins: @@ -145,7 +202,7 @@ DB-backed; defaults are in `BackgroundTasksSettings`. Several are marked `requir | `task_default_queue` | `"default"` | restart required | | `task_always_eager` | `False` | runs tasks in-process (test mode) | | `task_eager_propagates` | `True` | re-raises eager-mode errors | -| `stuck_after_seconds` | `300` | heartbeat-staleness threshold | +| `stuck_after_seconds` | `300` | heartbeat-staleness threshold; a task's own `stuck_after` overrides it — see [Long-running tasks](#long-running-tasks-heartbeats-and-stuck-thresholds) | | `stuck_sweep_interval_seconds` | `60` | beat cadence for the stuck sweep | | `purge_interval_seconds` | `86_400` | beat cadence for old-row purge | | `retention_days` | `14` | how long to keep terminal rows; range 1-3650 | @@ -170,7 +227,7 @@ Bootstrap env-var equivalents (`SM_BG_TASKS_*`) only seed pydantic defaults at f | Task | Schedule | What | |---|---|---| -| `background_tasks.sweep_stuck_tasks` | every `stuck_sweep_interval_seconds` | flips `running` rows with stale heartbeats to `stuck` | +| `background_tasks.sweep_stuck_tasks` | every `stuck_sweep_interval_seconds` | flips `running` rows whose heartbeat is older than their task's `stuck_after` (else `stuck_after_seconds`) to `stuck` | | `background_tasks.purge_old_executions` | every `purge_interval_seconds` | deletes terminal rows older than `retention_days` | A `background_tasks.demo_echo` task is also registered for smoke tests; not scheduled. @@ -202,7 +259,7 @@ Synchronous handlers in `signals.py`, all bound through Celery's `signals. |---|---| | `before_task_publish` | inserts a `pending` row | | `task_prerun` | flips to `running`, stamps `started_at` + `heartbeat_at` | -| `task_postrun` | refreshes `heartbeat_at` | +| `task_postrun` | refreshes `heartbeat_at` (between prerun and postrun, only `heartbeat()` from the task body does) | | `task_success` | flips to `success`, stores `result`, sets `finished_at` | | `task_failure` | flips to `failed`, stores `traceback` + `exception_type`, publishes `TaskFailed` | | `task_retry` | flips to `retrying`, increments `retries` | diff --git a/docs/modules/file_storage.md b/docs/modules/file_storage.md index 1907e89a..7c9c6dbf 100644 --- a/docs/modules/file_storage.md +++ b/docs/modules/file_storage.md @@ -22,7 +22,7 @@ Pluggable file storage with two shipped backends — local filesystem and S3-com | `GET /api/file-storage/files/{file_id}` | → `StoredFileOut` | `file_storage.download` | | `PATCH /api/file-storage/files/{file_id}` | `{"public": bool}` → `StoredFileOut` | `file_storage.upload` | | `GET /api/file-storage/files/{file_id}/thumbnail` | `?w=` → `image/webp` | `file_storage.download` | -| `GET /api/file-storage/files/{file_id}/download` | → 302 (S3) or stream (filesystem) | `file_storage.download` | +| `GET /api/file-storage/files/{file_id}/download` | → 302 (S3) or stream (filesystem); honours `Range` (see [Range requests](#range-requests)) | `file_storage.download` | | `DELETE /api/file-storage/files/{file_id}` | → 204 | `file_storage.delete` | | `GET /api/file-storage/public/{file_id}[/{filename}]` | bytes (**anonymous**) | none, `public` files only | | `GET /api/file-storage/public/{file_id}/thumbnail` | `?w=` → `image/webp` (**anonymous**) | none, `public` files only | @@ -85,6 +85,29 @@ Serving rules: - Presigning backends (S3) answer with a `302` to the presigned URL, cached for at most half the signature's TTL; filesystem backends stream. +### Range requests + +Whenever bytes stream through the app (the download endpoint on a +non-presigning backend, and the public route for filesystem backends or active +content), a single `Range` is honoured so COG / PMTiles readers can fetch +just the tiles they need: + +- `bytes=a-b`, `bytes=a-` and the suffix form `bytes=-n` answer `206 Partial + Content` with `Content-Range: bytes a-b/size` and the slice's + `Content-Length`. An end past the object is clamped to `size - 1`. +- A range starting at or past the end (or `bytes=-0`) is `416` with + `Content-Range: bytes */size`. +- Multiple ranges, another unit, or a malformed header are ignored and the + whole file is sent with `200` (RFC 9110 allows this). So is a request whose + `If-Range` does not match the current `ETag`. +- Streamed responses advertise `Accept-Ranges: bytes`. Auth, `ETag`, + `Content-Disposition`, `Cache-Control` and CSP headers are the same as for + the full response. + +Redirected (presigned) downloads are served by the object store, which handles +`Range` itself. Always redirecting large downloads to a presigned URL, as an +opt-in setting, is not implemented. + ### View | Method + path | Inertia component | Permission | @@ -229,12 +252,19 @@ class GcsBackend(StorageBackend): supports_presigned_url = True async def put(self, key, stream, content_type, size): ... - async def get(self, key): ... + async def get(self, key): ... # awaited; returns an AsyncIterator[bytes] + async def get_range(self, key, start, end): ... # end inclusive async def delete(self, key): ... async def exists(self, key) -> bool: ... async def presigned_get_url(self, key, ttl_seconds: int) -> str: ... ``` +`get` and `get_range` must stream in bounded chunks (never buffer the whole +object) and raise `StorageNotFoundError` from the awaited call, before the +first byte. `get_range(key, start, end)` uses HTTP semantics — `end` is +inclusive — and receives an already-validated `0 <= start <= end < size`. A +backend without `get_range` still works: the endpoints then ignore `Range`. + Register it in your module's `register_settings` so the file_storage service can pick it up. (The shipped two backends are wired into a private `BACKEND_REGISTRY`; see `backends/__init__.py` for the pattern.) ## Permissions diff --git a/docs/modules/users.md b/docs/modules/users.md index 1b3eb519..ddb982d6 100644 --- a/docs/modules/users.md +++ b/docs/modules/users.md @@ -384,6 +384,17 @@ Built-in provider keys (the `{provider}` URL segment): A single dispatcher pair (`/api/users/auth/{provider}/login` + `/callback`) serves every provider; the client is resolved per request from the cache. The `/callback` returns a 303 redirect (not the stock fastapi-users 204) so Inertia lands on a real page, with the auth cookie attached. Find-or-create + email association goes through `UserManager.oauth_callback` (`associate_by_email=True`, `is_verified_by_default=True`); state CSRF uses the signed session cookie. Linked accounts are stored in `users_oauth_account`. A **newly provisioned** OAuth account is marked external (see below); an existing password account linked by email keeps its password and is left unchanged. +## Importing users from another system + +`users` stores passwords through fastapi-users' default `PasswordHelper`, so you can import `hashed_password` values from another system and users keep their existing passwords (no reset needed). Accepted formats: + +| Format | Prefix | Role | +|---|---|---| +| Argon2 (`argon2id`/`argon2i`/`argon2d`, PHC string) | `$argon2...` | Current scheme, used for all new hashes | +| bcrypt | `$2a$` / `$2b$` / `$2y$` | Accepted on import (deprecated) | + +Other formats (PBKDF2, scrypt, SHA-based, plain text) are **not** recognised, and such users cannot sign in until they reset their password. A bcrypt hash is verified as-is and transparently re-hashed to Argon2 on the user's next successful sign-in, through both the session login (`/api/users/auth/login`) and the bearer-token login (`/api/users/auth/token`). Leave `hashed_password` as `NULL` for accounts that should only sign in through an external IdP (see below). + ## External / SSO users Users created through an external IdP (Google, GitHub, Microsoft/Entra ID, or generic OIDC) are provisioned with `is_external=True` and **no local password** (`hashed_password` is `NULL`). The OAuth callback's find-or-create only nulls the password + marks external for accounts it *creates* — an existing password account that gets linked by email keeps its password and stays non-external. External users sign in **only** through their IdP; roles are still assigned locally like any other user. diff --git a/docs/reference/make-commands.md b/docs/reference/make-commands.md index 85e2a425..d90b00f4 100644 --- a/docs/reference/make-commands.md +++ b/docs/reference/make-commands.md @@ -59,6 +59,7 @@ When a module declares a `simple_module_cli.cli_plugins` entry point, the CLI mo | Command | From | What | |---|---|---| | `smpy host gen-pages` | `simple_module_hosting` | Regenerate `client_app/modules.{manifest.json,generated.ts,generated.css}` from the installed modules. Run when you add or remove pages. | +| `smpy host gen-i18n` | `simple_module_hosting` | Regenerate the typed i18n keys (`keys.generated.ts`) into the installed `@simple-module-py/i18n` package (or `packages/i18n/src` in the framework repo) without booting the app. Fails if neither exists. | | `smpy host sync-js-deps` | `simple_module_hosting` | Install JS deps declared by wheel-installed modules into `client_app/node_modules`. In-repo workspace modules don't need this. | | `smpy settings import-from-env` | `simple_module_settings` | One-shot migration: read every `SM__*` env var and seed it as a SYSTEM-scope row in the DB-backed settings store. Idempotent. | @@ -80,6 +81,8 @@ uv run smpy users create-admin --email admin@example.com --password changeme | `make build` | `cd client_app && npm run build`. Produces the production frontend bundle. | | `make gen-pages` | Regenerate the page manifest. Auto-runs before `make dev`. Same effect as `smpy host gen-pages`. | | `make sync-js-deps` | Install JS deps from wheel-installed modules. Same effect as `smpy host sync-js-deps`. | +| `make gen-i18n` | Regenerate the typed `t(keys....)` union in `node_modules/@simple-module-py/i18n/src`. Dev boot does this too; `make install` runs it once so `tsc` works before the first boot. Same effect as `smpy host gen-i18n`. | +| `make kill` | Free the dev ports with `lsof`, nothing else: `API_PORT` (default `8000`) and `VITE_PORT` (read from `SM_VITE_DEV_URL` in `.env`, default `5050`). Never pkills by process name, so other apps' servers and test runs survive. Override with `make dev API_PORT=8010`. | If you included `background_tasks` in `smpy new`, the scaffold appends `worker`, `beat`, and `worker-docker` targets that wrap the Celery commands — read your `Makefile` to see the exact invocations. @@ -98,7 +101,7 @@ The scaffolded Makefile intentionally doesn't wrap one-off things. Run them dire | Type-check Python | `uv run ty check` | | Format + lint JS | `cd client_app && npx biome check .` | | Type-check JS | `cd client_app && npx tsc --noEmit` | -| Free stuck dev ports | `lsof -ti:8000,5050 \| xargs kill -9` | +| Free stuck dev ports | `make kill` (frees only `API_PORT`/`VITE_PORT`) | ## Working on the framework itself diff --git a/framework/cli/simple_module_cli/_env.py b/framework/cli/simple_module_cli/_env.py index 16e7e11b..a2184c15 100644 --- a/framework/cli/simple_module_cli/_env.py +++ b/framework/cli/simple_module_cli/_env.py @@ -15,4 +15,5 @@ def set_env_key(text: str, key: str, value: str) -> str: def env_values(text: str) -> dict[str, str]: """Parse the ``KEY=VALUE`` lines of an env-style file body (comments skipped).""" lines = text.splitlines() - return dict(ln.split("=", 1) for ln in lines if "=" in ln and not ln.startswith("#")) + pairs = (ln.split("=", 1) for ln in lines if "=" in ln and not ln.lstrip().startswith("#")) + return {key.strip(): value for key, value in pairs} diff --git a/framework/cli/simple_module_cli/_local_env.py b/framework/cli/simple_module_cli/_local_env.py index a083a809..891f9d4a 100644 --- a/framework/cli/simple_module_cli/_local_env.py +++ b/framework/cli/simple_module_cli/_local_env.py @@ -14,7 +14,12 @@ from simple_module_cli._env import env_values, set_env_key -__all__ = ["comment_out_env_key", "write_local_env"] +__all__ = ["add_postgres_password_placeholder", "comment_out_env_key", "write_local_env"] + +_PG_PASSWORD_NOTE = ( + "# Compose puts this into SM_DATABASE_URL unencoded: use only URL-safe\n" + "# characters (letters, digits, - _ . ~). `make docker-up` rejects others (GH #472).\n" +) _SECRET_NOTE = ( "# Secrets — `smpy new` generated these into .env (gitignored); keep them\n" @@ -37,6 +42,12 @@ def comment_out_env_key(text: str, key: str) -> str: return "\n".join(lines) + "\n" +def add_postgres_password_placeholder(text: str) -> str: + """Add an empty ``POSTGRES_PASSWORD=`` to ``.env.example``, with its URL-safe note.""" + text = set_env_key(text, "POSTGRES_PASSWORD", "") + return text.replace("POSTGRES_PASSWORD=\n", _PG_PASSWORD_NOTE + "POSTGRES_PASSWORD=\n", 1) + + def write_local_env( target: Path, *, secret_keys: Iterable[str], overrides: dict[str, str] | None = None ) -> None: diff --git a/framework/cli/simple_module_cli/app_project.py b/framework/cli/simple_module_cli/app_project.py index 8a64e3fa..301c7fc5 100644 --- a/framework/cli/simple_module_cli/app_project.py +++ b/framework/cli/simple_module_cli/app_project.py @@ -20,7 +20,11 @@ from typing import Any from simple_module_cli._env import set_env_key -from simple_module_cli._local_env import comment_out_env_key, write_local_env +from simple_module_cli._local_env import ( + add_postgres_password_placeholder, + comment_out_env_key, + write_local_env, +) from simple_module_cli.case import to_kebab_case, to_pascal_case, to_snake_case from simple_module_cli.catalog import CATALOG, PRESETS, expand_deps from simple_module_cli.docker_assets import scaffold_docker_assets @@ -173,7 +177,7 @@ def create_app_project( if db == "postgres": # The compose `postgres` service takes its password from .env, and # the dev SM_DATABASE_URL must match it (GH #444). - env_text = set_env_key(env_text, "POSTGRES_PASSWORD", "") + env_text = add_postgres_password_placeholder(env_text) password = _secrets.token_urlsafe(24) local_overrides = { "POSTGRES_PASSWORD": password, diff --git a/framework/cli/simple_module_cli/cli.py b/framework/cli/simple_module_cli/cli.py index 6a4aabb8..aee9d248 100644 --- a/framework/cli/simple_module_cli/cli.py +++ b/framework/cli/simple_module_cli/cli.py @@ -114,6 +114,8 @@ def _echo_workspace_hint(target: Path) -> None: if not (host / "pyproject.toml").is_file() or not is_workspace_member(host, target): return rel = os.path.relpath(target.resolve(), host.resolve()) + if not rel.startswith("."): + rel = f"./{rel}" # parse_add_spec reads a bare "a/b" as a PyPI requirement host_arg = os.path.relpath(host.resolve(), Path.cwd()) if host_arg.startswith(".."): host_arg = str(host.resolve()) diff --git a/framework/cli/simple_module_cli/templates/docker/Makefile.snippet b/framework/cli/simple_module_cli/templates/docker/Makefile.snippet index 6a8330bc..7544f42b 100644 --- a/framework/cli/simple_module_cli/templates/docker/Makefile.snippet +++ b/framework/cli/simple_module_cli/templates/docker/Makefile.snippet @@ -8,6 +8,16 @@ docker-up: ## Run the full stack in containers @test -f .env || { cp .env.example .env; \ echo "Created .env from .env.example: set POSTGRES_PASSWORD (Postgres apps) and the"; \ echo "commented SM_* secrets in .env; production containers refuse to boot without them."; } + @# compose interpolates POSTGRES_PASSWORD raw into the SM_DATABASE_URL, so it must be URL-safe. + @# The value is read the way compose reads it: last definition wins, CRLF, an inline + @# trailing comment and surrounding quotes are not part of the password. + @pw=$$(sed -n 's/^POSTGRES_PASSWORD=//p' .env | tail -n 1 | tr -d '\r' \ + | sed -e 's/[[:space:]]#.*$$//' -e 's/[[:space:]]*$$//' -e 's/^"\(.*\)"$$/\1/' -e "s/^'\(.*\)'$$/\1/"); \ + if printf '%s' "$$pw" | grep -q '[^A-Za-z0-9._~-]'; then \ + echo "POSTGRES_PASSWORD in .env must be URL-safe (letters, digits, - _ . ~ only):" >&2; \ + echo "compose puts it into SM_DATABASE_URL unencoded. Pick another password." >&2; \ + exit 1; \ + fi docker compose up --build docker-down: ## Stop the compose stack diff --git a/framework/cli/simple_module_cli/templates/host/Makefile b/framework/cli/simple_module_cli/templates/host/Makefile index dce73537..0ad61ae5 100644 --- a/framework/cli/simple_module_cli/templates/host/Makefile +++ b/framework/cli/simple_module_cli/templates/host/Makefile @@ -1,16 +1,24 @@ -.PHONY: install dev dev-api dev-ui build test test-py test-js lint doctor migrate migration gen-pages sync-js-deps +.PHONY: install dev dev-api dev-ui build test test-py test-js lint doctor migrate migration gen-pages sync-js-deps kill gen-i18n + +# Dev ports. VITE_PORT follows SM_VITE_DEV_URL in .env (the same value Vite and +# the backend read), falling back to 5050. Override on the command line: +# `make dev API_PORT=8010`. `make kill` frees only these two ports, so other +# apps' dev servers (and vitest runs) on the machine are left alone. +API_PORT ?= 8000 +VITE_PORT ?= $(or $(shell sed -n 's/^SM_VITE_DEV_URL=.*:\([0-9][0-9]*\)\/*$$/\1/p' .env 2>/dev/null | tail -n 1),5050) install: uv sync cd client_app && npm install $(MAKE) sync-js-deps + $(MAKE) gen-i18n dev: gen-pages @echo "Starting API and UI dev servers..." $(MAKE) -j2 dev-api dev-ui dev-api: - uv run uvicorn main:app --reload --port 8000 + uv run uvicorn main:app --reload --port $(API_PORT) dev-ui: cd client_app && npm run dev @@ -45,6 +53,13 @@ gen-pages: sync-js-deps: uv run python -m simple_module_hosting sync-js-deps --host-client-app=client_app +# Regenerate the typed i18n keys (`t(keys....)`) into the installed +# @simple-module-py/i18n package. Dev boot does this too; run it after editing a +# locale catalog, before `tsc`, or in CI where nothing boots the app. Fails if +# the package is not installed (run `npm install` first). +gen-i18n: + uv run python -m simple_module_hosting gen-i18n + # `upgrade heads` (plural) applies every per-module branch head; `upgrade head` # (singular) errors once a second module adds its own migration branch label. migrate: @@ -53,3 +68,7 @@ migrate: migration: @test -n "$(msg)" || (echo 'Usage: make migration msg="describe the change"' && exit 1) uv run alembic revision --autogenerate -m "$(msg)" + +kill: + @-lsof -ti:$(API_PORT),$(VITE_PORT) | xargs -r kill 2>/dev/null + @echo "Ports $(API_PORT), $(VITE_PORT) freed." diff --git a/framework/cli/simple_module_cli/templates/workspace/Makefile b/framework/cli/simple_module_cli/templates/workspace/Makefile index 7db00dad..edaa9f89 100644 --- a/framework/cli/simple_module_cli/templates/workspace/Makefile +++ b/framework/cli/simple_module_cli/templates/workspace/Makefile @@ -1,9 +1,17 @@ -.PHONY: install dev dev-api dev-ui build test test-py test-js lint doctor migrate migration gen-pages sync-module-deps kill +.PHONY: install dev dev-api dev-ui build test test-py test-js lint doctor migrate migration gen-pages sync-module-deps kill gen-i18n + +# Dev ports. VITE_PORT follows SM_VITE_DEV_URL in .env (the same value Vite and +# the backend read), falling back to 5050. Override on the command line: +# `make dev API_PORT=8010`. `make kill` frees only these two ports, so other +# apps' dev servers (and vitest runs) on the machine are left alone. +API_PORT ?= 8000 +VITE_PORT ?= $(or $(shell sed -n 's/^SM_VITE_DEV_URL=.*:\([0-9][0-9]*\)\/*$$/\1/p' .env 2>/dev/null | tail -n 1),5050) install: uv sync --all-packages npm install $(MAKE) sync-module-deps + $(MAKE) gen-i18n dev: gen-pages @echo "Starting API and UI dev servers..." @@ -13,7 +21,7 @@ dev: gen-pages # modules/* as well as the host, so edits to a module's routes/endpoints/ # locales hot-reload too (uvicorn otherwise watches only the launch cwd). dev-api: - cd host && uv run uvicorn main:app --reload --reload-dir . --reload-dir ../modules --port 8000 + cd host && uv run uvicorn main:app --reload --reload-dir . --reload-dir ../modules --port $(API_PORT) dev-ui: npm run dev @@ -52,6 +60,13 @@ gen-pages: sync-module-deps: cd host && uv run python -m simple_module_hosting sync-js-deps --host-client-app=client_app +# Regenerate the typed i18n keys (`t(keys....)`) into the installed +# @simple-module-py/i18n package. Dev boot does this too; run it after editing a +# locale catalog, before `tsc`, or in CI where nothing boots the app. Fails if +# the package is not installed (run `npm install` first). +gen-i18n: + cd host && uv run python -m simple_module_hosting gen-i18n + # `upgrade heads` (plural) applies every per-module branch head; `upgrade head` # (singular) errors once a second module adds its own migration branch label. # @@ -68,7 +83,5 @@ migration: uv run --project host alembic -c host/alembic.ini revision --autogenerate -m "$(msg)" kill: - @-pkill -f "uvicorn main:app" 2>/dev/null - @-pkill -f vite 2>/dev/null - @-lsof -ti:8000,5050 | xargs kill -9 2>/dev/null - @echo "Ports 8000, 5050 freed." + @-lsof -ti:$(API_PORT),$(VITE_PORT) | xargs -r kill 2>/dev/null + @echo "Ports $(API_PORT), $(VITE_PORT) freed." diff --git a/framework/cli/tests/test_cli_add_workspace_member.py b/framework/cli/tests/test_cli_add_workspace_member.py index c6df0b17..17c74868 100644 --- a/framework/cli/tests/test_cli_add_workspace_member.py +++ b/framework/cli/tests/test_cli_add_workspace_member.py @@ -8,6 +8,8 @@ import pytest import typer from simple_module_cli.add_cmd import run_add +from simple_module_cli.cli import _echo_workspace_hint +from simple_module_cli.git_source import parse_add_spec from simple_module_cli.uv_workspace import is_workspace_member HOST = '[project]\nname = "myhost"\nversion = "0.1.0"\ndependencies = []\n' @@ -87,3 +89,24 @@ def test_missing_path_error_names_the_resolution_base( exec_runner=_noop, ) assert "--path project" in capsys.readouterr().err + + +def test_workspace_hint_inside_host_dir_prints_a_path_spec( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + # GH #471: a bare "modules/x" would be parsed by `smpy add` as a PyPI requirement. + (tmp_path / "pyproject.toml").write_text( + '[tool.uv.workspace]\nmembers = ["host", "host/modules/*"]\n', encoding="utf-8" + ) + (tmp_path / "host").mkdir() + (tmp_path / "host" / "pyproject.toml").write_text(HOST, encoding="utf-8") + module = tmp_path / "host" / "modules" / "forest_plots" + _write_module(module) + monkeypatch.chdir(tmp_path) + + _echo_workspace_hint(module) + + out = capsys.readouterr().out + spec = out.split("smpy add ", 1)[1].split(" --path", 1)[0] + assert spec == "./modules/forest_plots" + assert parse_add_spec(spec).kind == "path" diff --git a/framework/cli/tests/test_cli_docker_assets.py b/framework/cli/tests/test_cli_docker_assets.py index f5cefde2..0e2a2d65 100644 --- a/framework/cli/tests/test_cli_docker_assets.py +++ b/framework/cli/tests/test_cli_docker_assets.py @@ -2,6 +2,9 @@ from __future__ import annotations +import os +import shutil +import subprocess from pathlib import Path import pytest @@ -90,6 +93,52 @@ def test_postgres_scaffold_env_matches_compose_password(tmp_path: Path) -> None: assert "POSTGRES_PASSWORD=\n" in example +def test_postgres_password_must_be_url_safe_is_documented_and_enforced(tmp_path: Path) -> None: + # GH #472: compose cannot URL-encode, so the raw password lands in the URL. + target = tmp_path / "demo" + _create(target, db="postgres") + example = (target / ".env.example").read_text() + assert "URL-safe" in example + assert example.index("URL-safe") < example.index("POSTGRES_PASSWORD=") + makefile = (target / "Makefile").read_text() + assert makefile.index("URL-safe") < makefile.index("docker compose up") + + +@pytest.mark.skipif(shutil.which("make") is None, reason="needs make") +@pytest.mark.parametrize( + ("env_body", "ok"), + [ + ("POSTGRES_PASSWORD=abc-_.~123\n", True), + ('POSTGRES_PASSWORD="abc123"\n', True), + ("POSTGRES_PASSWORD='abc123'\n", True), + ("POSTGRES_PASSWORD=abc123\r\n", True), + ("POSTGRES_PASSWORD=abc123 # local only\n", True), + ("POSTGRES_PASSWORD=\n", True), + ("# POSTGRES_PASSWORD=a/b\nPOSTGRES_PASSWORD=ok\n", True), + ("POSTGRES_PASSWORD=a/b@c\n", False), + ('POSTGRES_PASSWORD="a:b"\n', False), + ("POSTGRES_PASSWORD=ok\nPOSTGRES_PASSWORD=p#ss word\n", False), + ], +) +def test_docker_up_guard_reads_password_like_compose( + tmp_path: Path, env_body: str, ok: bool +) -> None: + target = tmp_path / "demo" + _create(target, db="postgres") + (target / ".env").write_bytes(env_body.encode()) + # Stub `docker` so a passing guard does not start anything. + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + stub = bin_dir / "docker" + stub.write_text("#!/bin/sh\nexit 0\n") + stub.chmod(0o755) + env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} + result = subprocess.run( + ["make", "docker-up"], cwd=target, env=env, capture_output=True, text=True, check=False + ) + assert (result.returncode == 0) is ok, result.stderr + + def test_compose_containers_run_production_env(tmp_path: Path) -> None: # Development mode emits asset tags pointing at the (absent) Vite dev # server, so containers must boot in production mode. diff --git a/framework/cli/tests/test_cli_new.py b/framework/cli/tests/test_cli_new.py index a919eb54..c3eacfe2 100644 --- a/framework/cli/tests/test_cli_new.py +++ b/framework/cli/tests/test_cli_new.py @@ -263,3 +263,8 @@ def test_write_local_env_keeps_existing_postgres_password_and_url_together( overrides={"POSTGRES_PASSWORD": "new", "SM_DATABASE_URL": "postgresql://postgres:new@x/db"}, ) assert env_values((tmp_path / ".env").read_text()) == {"POSTGRES_PASSWORD": "kept"} + + +def test_env_values_ignores_indented_comments_and_strips_key_spacing() -> None: + text = " # SM_SECRET_KEY=nope\nA = 1\n# B=2\nC=3\n" + assert env_values(text) == {"A": " 1", "C": "3"} diff --git a/framework/cli/tests/test_cli_new_scaffold_layout.py b/framework/cli/tests/test_cli_new_scaffold_layout.py index 2af9f596..7416f3aa 100644 --- a/framework/cli/tests/test_cli_new_scaffold_layout.py +++ b/framework/cli/tests/test_cli_new_scaffold_layout.py @@ -128,6 +128,38 @@ def test_sm_new_makefile_has_quality_gate_targets(tmp_path: Path) -> None: assert "upgrade head\n" not in makefile # the buggy singular form is gone +def test_sm_new_makefile_kill_is_port_scoped(tmp_path: Path) -> None: + """Regression #461: `make kill` must free only this app's ports, never + pkill every uvicorn/vite (other apps, vitest) on the machine.""" + from simple_module_cli.app_project import create_app_project + + target = tmp_path / "demo" + create_app_project(target, name="demo", db="sqlite", tenancy=False, selected=[]) + makefile = (target / "Makefile").read_text() + assert "pkill" not in makefile + assert "lsof -ti:$(API_PORT),$(VITE_PORT)" in makefile + assert "--port $(API_PORT)" in makefile + assert "SM_VITE_DEV_URL" in makefile + + +def test_sm_new_makefile_has_gen_i18n_target(tmp_path: Path) -> None: + """Regression #449: scaffolded apps regenerate the typed i18n keys on demand.""" + from simple_module_cli.app_project import create_app_project + + target = tmp_path / "demo" + create_app_project(target, name="demo", db="sqlite", tenancy=False, selected=[]) + makefile = (target / "Makefile").read_text() + assert "gen-i18n:" in makefile + assert "simple_module_hosting gen-i18n" in makefile + + +def test_host_template_makefile_has_no_pkill() -> None: + from simple_module_cli import app_project + + host_makefile = Path(app_project.__file__).parent / "templates" / "host" / "Makefile" + assert "pkill" not in host_makefile.read_text() + + def test_sm_new_root_pyproject_has_dev_tooling_and_config(tmp_path: Path) -> None: """Regression #201: the workspace root must ship the dev tooling + pytest / ruff / ty config so `make test` and `make lint` work out of the box.""" diff --git a/framework/db/simple_module_db/__init__.py b/framework/db/simple_module_db/__init__.py index 61e64b22..478d6320 100644 --- a/framework/db/simple_module_db/__init__.py +++ b/framework/db/simple_module_db/__init__.py @@ -29,6 +29,7 @@ tenant_context, ) from simple_module_db.transaction import CommitBeforeResponseMiddleware, finalize_session +from simple_module_db.types import AwareDateTime from simple_module_db.writes import hard_delete, mark_written __all__ = [ @@ -39,6 +40,7 @@ "TENANT_ID_PATTERN", "AuditMixin", "AuditRecord", + "AwareDateTime", "CommitBeforeResponseMiddleware", "DatabaseProvider", "DatabaseState", diff --git a/framework/db/simple_module_db/migrations.py b/framework/db/simple_module_db/migrations.py index fbefe033..0ab3f6ed 100644 --- a/framework/db/simple_module_db/migrations.py +++ b/framework/db/simple_module_db/migrations.py @@ -27,6 +27,7 @@ from simple_module_core.discovery import discover_modules, get_module_package_name from sqlalchemy import Index, MetaData from sqlalchemy.schema import SchemaItem +from sqlmodel.sql.sqltypes import UTCDateTime from simple_module_db.base import all_module_bases from simple_module_db.migration_portability import ( @@ -259,7 +260,7 @@ def render_item(type_, obj, autogen_context): if cls_name == "AutoString": length = getattr(obj, "length", None) return f"sa.String(length={length})" if length is not None else "sa.String()" - if cls_name == "UTCDateTime" and cls_module.startswith("sqlmodel"): + if isinstance(obj, UTCDateTime): # includes our ``AwareDateTime`` subclass return "sa.DateTime(timezone=True)" if imports is not None and cls_module.startswith("sqlmodel"): diff --git a/framework/db/simple_module_db/mixins.py b/framework/db/simple_module_db/mixins.py index 3db76f87..a807f615 100644 --- a/framework/db/simple_module_db/mixins.py +++ b/framework/db/simple_module_db/mixins.py @@ -10,9 +10,11 @@ from datetime import UTC, datetime -from sqlalchemy import DateTime, func +from sqlalchemy import func from sqlmodel import Field, SQLModel +from simple_module_db.types import AwareDateTime + def _utcnow() -> datetime: """Timezone-aware UTC ``now`` for the Python-side default of ``created_at``.""" @@ -32,12 +34,12 @@ class AuditMixin(SQLModel): created_at: datetime = Field( default_factory=_utcnow, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), sa_column_kwargs={"server_default": func.now()}, ) updated_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), sa_column_kwargs={"onupdate": func.now()}, ) created_by: str | None = Field(default=None, max_length=255) @@ -55,7 +57,7 @@ class SoftDeleteMixin(SQLModel): is_deleted: bool = Field(default=False) deleted_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), ) deleted_by: str | None = Field(default=None, max_length=255) diff --git a/framework/db/simple_module_db/types.py b/framework/db/simple_module_db/types.py new file mode 100644 index 00000000..135597ef --- /dev/null +++ b/framework/db/simple_module_db/types.py @@ -0,0 +1,39 @@ +"""Column types shared by every module. + +``AwareDateTime`` is the explicit-column counterpart of the ``UTCDateTime`` +SQLModel picks for a bare ``datetime`` field. Declaring +``sa_type=DateTime(timezone=True)`` bypasses that type, and on SQLite (which +stores no offset) such a column reads back *naive*: pydantic then serializes +``2026-10-14T10:40:52`` with no offset and a browser parses UTC as local time +(GH #468). Use ``sa_type=AwareDateTime()`` wherever a field needs an explicit +column type. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy.engine.interfaces import Dialect +from sqlmodel.sql.sqltypes import UTCDateTime + + +class AwareDateTime(UTCDateTime): + """``DateTime(timezone=True)`` that always reads back an aware UTC value. + + Unlike SQLModel's ``UTCDateTime`` a naive value is accepted on write and + taken to be UTC — the semantics a plain ``DateTime(timezone=True)`` column + had on SQLite — so existing callers that bind naive UTC keep working. + Migrations render it as ``sa.DateTime(timezone=True)`` (see + :func:`simple_module_db.migrations.render_item`), so adopting it needs no + schema change. + """ + + cache_ok = True + + def __repr__(self) -> str: + return "AwareDateTime()" + + def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None: + if value is not None and value.utcoffset() is None: + value = value.replace(tzinfo=UTC) + return super().process_bind_param(value, dialect) diff --git a/framework/db/tests/test_aware_datetime.py b/framework/db/tests/test_aware_datetime.py new file mode 100644 index 00000000..2e6ee1d4 --- /dev/null +++ b/framework/db/tests/test_aware_datetime.py @@ -0,0 +1,76 @@ +"""GH #468: timezone-aware columns must read back aware on SQLite. + +SQLite stores no offset, so a plain ``DateTime(timezone=True)`` column reads +back naive and pydantic serializes it without one — the browser then parses +UTC as local time. ``AwareDateTime`` attaches UTC on read. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta, timezone +from types import SimpleNamespace + +import sqlalchemy as sa +from simple_module_db import AuditMixin, AwareDateTime, render_item +from sqlalchemy.ext.asyncio import create_async_engine + +_md = sa.MetaData() +_stamps = sa.Table( + "aware_dt_stamps", + _md, + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("at", AwareDateTime()), +) + + +async def _roundtrip(value: datetime) -> datetime: + engine = create_async_engine("sqlite+aiosqlite://") + try: + async with engine.begin() as conn: + await conn.run_sync(_md.create_all) + await conn.execute(_stamps.insert().values(id=1, at=value)) + return (await conn.execute(sa.select(_stamps.c.at))).scalar_one() + finally: + await engine.dispose() + + +async def test_aware_value_reads_back_aware_utc(): + stamp = datetime(2026, 10, 14, 10, 40, 52, 123456, tzinfo=UTC) + + loaded = await _roundtrip(stamp) + + assert loaded == stamp + assert loaded.utcoffset() == timedelta(0) + assert loaded.isoformat().endswith("+00:00") + + +async def test_non_utc_offset_is_normalised_to_utc(): + stamp = datetime(2026, 10, 14, 12, 40, tzinfo=timezone(timedelta(hours=2))) + + loaded = await _roundtrip(stamp) + + assert loaded == stamp + assert loaded.tzinfo == UTC + assert loaded.hour == 10 + + +async def test_naive_value_is_taken_as_utc(): + """Unlike SQLModel's ``UTCDateTime``, a naive bind is accepted as UTC.""" + loaded = await _roundtrip(datetime(2026, 10, 14, 10, 40)) + + assert loaded == datetime(2026, 10, 14, 10, 40, tzinfo=UTC) + + +def test_audit_mixin_columns_use_aware_type(): + fields = AuditMixin.model_fields + for name in ("created_at", "updated_at"): + sa_type = getattr(fields[name], "sa_type", None) + assert isinstance(sa_type, AwareDateTime), name + + +def test_renders_as_plain_timezone_datetime(): + """Adopting the type must not make autogenerate emit a diff or a sqlmodel import.""" + ctx = SimpleNamespace(imports=set()) + + assert render_item("type", AwareDateTime(), ctx) == "sa.DateTime(timezone=True)" + assert ctx.imports == set() diff --git a/framework/hosting/simple_module_hosting/_i18n_target.py b/framework/hosting/simple_module_hosting/_i18n_target.py new file mode 100644 index 00000000..782d2d67 --- /dev/null +++ b/framework/hosting/simple_module_hosting/_i18n_target.py @@ -0,0 +1,40 @@ +"""Where generated i18n key files are written: monorepo checkout or published npm package.""" + +from __future__ import annotations + +import re +from pathlib import Path + +RESOURCE_KEY_RE = re.compile(r"^\s+'([^']+)': '',$", re.MULTILINE) + +_PUBLISHED_PKG = Path("node_modules") / "@simple-module-py" / "i18n" / "src" + + +def resolve_i18n_types_dir(project_root: Path) -> tuple[Path | None, bool]: + """Where the generated key files go, and whether that is a published package. + + The monorepo's ``packages/i18n/src`` wins. A scaffolded app has no such + checkout — ``@simple-module-py/i18n`` is an npm dependency — so walk up from + ``project_root`` to the nearest ``node_modules`` holding it (npm workspaces + hoist it to the workspace root, above the host; a flat host installs into + ``client_app/``). ``(None, False)`` when + neither exists, the normal shape of a wheel-installed prod app. + """ + local = project_root / "packages" / "i18n" / "src" + if local.is_dir(): + return local, False + root = project_root.resolve() + # A flat host (``smpy new --flat``) keeps its npm install in client_app/. + for base in (root, root / "client_app", *root.parents): + published = base / _PUBLISHED_PKG + if published.is_dir(): + return published, True + return None, False + + +def read_resource_keys(resources_path: Path) -> frozenset[str]: + """Keys listed in an existing ``generated-resources.ts`` (empty if unreadable).""" + try: + return frozenset(RESOURCE_KEY_RE.findall(resources_path.read_text(encoding="utf-8"))) + except OSError: + return frozenset() diff --git a/framework/hosting/simple_module_hosting/host_cli.py b/framework/hosting/simple_module_hosting/host_cli.py index 64054af4..10681ca2 100644 --- a/framework/hosting/simple_module_hosting/host_cli.py +++ b/framework/hosting/simple_module_hosting/host_cli.py @@ -53,6 +53,47 @@ def gen_pages( ) +@app.command("gen-i18n") +def gen_i18n( + allow_removals: Annotated[ + bool, + typer.Option( + "--allow-removals", + help="Permit dropping key namespaces of modules not installed here.", + ), + ] = False, +) -> None: + """Regenerate the typed i18n key files (keys.generated.ts) without booting the app. + + Writes into packages/i18n/src in the monorepo, else into the installed + @simple-module-py/i18n package under node_modules. Fails loudly if neither + exists. + """ + from simple_module_hosting._project_root import resolve_project_root + from simple_module_hosting.i18n_manifest import ( + NamespaceRemovalError, + emit_frontend_types_for_modules, + ) + from simple_module_hosting.settings import Settings + + logging.basicConfig(level=logging.WARNING, format="%(message)s") + logging.getLogger("simple_module_hosting.i18n_manifest").setLevel(logging.INFO) + settings = Settings() + modules = discover_modules(enabled=settings.modules_enabled, strict=not settings.is_development) + try: + emit_frontend_types_for_modules( + settings, + modules, + resolve_project_root(), + strict=True, + allow_removals=allow_removals, + ) + except (NamespaceRemovalError, FileNotFoundError) as exc: + typer.echo(f"error: {exc}", err=True) + raise typer.Exit(code=1) from exc + typer.echo("i18n key files up to date") + + @app.command("sync-js-deps") def sync_js_deps( host_client_app: Annotated[ diff --git a/framework/hosting/simple_module_hosting/i18n_manifest.py b/framework/hosting/simple_module_hosting/i18n_manifest.py index 59a6d9b2..b4aef823 100644 --- a/framework/hosting/simple_module_hosting/i18n_manifest.py +++ b/framework/hosting/simple_module_hosting/i18n_manifest.py @@ -14,8 +14,8 @@ from __future__ import annotations import logging -import re from collections import Counter +from collections.abc import Iterable from pathlib import Path from typing import Any @@ -23,6 +23,7 @@ from simple_module_core.i18n import PLURAL_CATEGORIES, I18nRegistry from simple_module_hosting._i18n_build import build_i18n_registry +from simple_module_hosting._i18n_target import read_resource_keys, resolve_i18n_types_dir from simple_module_hosting.manifest import _write_if_changed from simple_module_hosting.settings import Settings @@ -77,14 +78,25 @@ def emit_frontend_types( skips the check, for a namespace that was deleted on purpose or a deliberate ``SM_MODULES_ENABLED`` subset. """ - pkg_src = project_root / "packages" / "i18n" / "src" - if not pkg_src.is_dir(): + pkg_src, published = resolve_i18n_types_dir(project_root) + if pkg_src is None: if strict: - raise FileNotFoundError(f"{pkg_src} does not exist — nothing to regenerate") + raise FileNotFoundError( + f"no @simple-module-py/i18n found for {project_root} (packages/i18n/src or " + "node_modules/@simple-module-py/i18n/src) — run `npm install` first" + ) return - if not allow_removals: + resources_path = pkg_src / "generated-resources.ts" + # In the monorepo the checked-in file is ours, so a vanished namespace is a + # mistake to guard against. A published package's file is the framework's + # full catalog: this app has fewer modules by design, so those keys are + # carried over instead of treated as removals. + extra_keys: frozenset[str] = frozenset() + if published: + extra_keys = read_resource_keys(resources_path) + elif not allow_removals: dropped = find_dropped_namespaces( - pkg_src / "generated-resources.ts", registry.messages(registry.default_locale) + resources_path, registry.messages(registry.default_locale) ) if dropped: error = NamespaceRemovalError(dropped) @@ -93,16 +105,13 @@ def emit_frontend_types( logger.warning("%s Not writing the generated i18n files.", error) return try: - write_generated_resources(registry, pkg_src) + write_generated_resources(registry, pkg_src, extra_keys=extra_keys) except Exception: if strict: raise logger.exception("Failed to write generated-resources.ts — frontend types will be stale") -_RESOURCE_KEY_RE = re.compile(r"^\s+'([^']+)': '',$", re.MULTILINE) - - class NamespaceRemovalError(RuntimeError): """Regenerating would delete whole key namespaces from the checked-in files.""" @@ -123,11 +132,7 @@ def find_dropped_namespaces(resources_path: Path, new_messages: Any) -> dict[str Returns ``{namespace: existing key count}``. Empty when the file is absent or unreadable (nothing to lose). """ - try: - existing = resources_path.read_text(encoding="utf-8") - except OSError: - return {} - old = Counter(key.split(".", 1)[0] for key in _RESOURCE_KEY_RE.findall(existing)) + old = Counter(key.split(".", 1)[0] for key in read_resource_keys(resources_path)) new = {key.split(".", 1)[0] for key in new_messages} return {ns: n for ns, n in old.items() if ns not in new} @@ -142,18 +147,21 @@ def find_dropped_namespaces(resources_path: Path, new_messages: Any) -> dict[str _PLURAL_SUFFIXES: tuple[str, ...] = tuple(f"_{c}" for c in PLURAL_CATEGORIES) -def write_generated_resources(registry: I18nRegistry, output_dir: Path) -> Path: +def write_generated_resources( + registry: I18nRegistry, output_dir: Path, *, extra_keys: Iterable[str] = () +) -> Path: """Write both ``generated-resources.ts`` and ``keys.generated.ts``. Returns the path of the resources file. Both files skip the disk write when content matches what's already on disk (to avoid bumping mtimes - that would trigger spurious Vite HMR). + that would trigger spurious Vite HMR). ``extra_keys`` are unioned into the + registry's keys (the published package's catalog, for a scaffolded app). """ output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) messages = registry.messages(registry.default_locale) - keys = sorted(messages.keys()) + keys = sorted({*messages.keys(), *extra_keys}) resources_path = output_dir / "generated-resources.ts" if _write_if_changed(resources_path, _render_resources(keys)): diff --git a/framework/hosting/tests/test_i18n_manifest_published_target.py b/framework/hosting/tests/test_i18n_manifest_published_target.py new file mode 100644 index 00000000..b487a1ee --- /dev/null +++ b/framework/hosting/tests/test_i18n_manifest_published_target.py @@ -0,0 +1,74 @@ +"""emit_frontend_types target resolution: monorepo vs published npm package vs neither.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from simple_module_core.i18n import I18nRegistry +from simple_module_hosting.i18n_manifest import ( + emit_frontend_types, + resolve_i18n_types_dir, +) + + +def _registry(*keys: str) -> I18nRegistry: + reg = I18nRegistry(default_locale="en", supported_locales=["en"]) + reg._messages = {"en": dict.fromkeys(keys, "x")} + return reg + + +def _published(root: Path) -> Path: + src = root / "node_modules" / "@simple-module-py" / "i18n" / "src" + src.mkdir(parents=True) + return src + + +def test_monorepo_checkout_wins_over_node_modules(tmp_path: Path) -> None: + local = tmp_path / "packages" / "i18n" / "src" + local.mkdir(parents=True) + _published(tmp_path) + assert resolve_i18n_types_dir(tmp_path) == (local, False) + + +def test_walks_up_to_hoisted_node_modules(tmp_path: Path) -> None: + src = _published(tmp_path) + host = tmp_path / "host" + host.mkdir() + found, published = resolve_i18n_types_dir(host) + assert published is True + assert found is not None and found.resolve() == src.resolve() + + +def test_neither_present_resolves_to_none(tmp_path: Path) -> None: + assert resolve_i18n_types_dir(tmp_path) == (None, False) + + +def test_boot_path_is_silent_noop_when_neither_exists(tmp_path: Path) -> None: + emit_frontend_types(_registry("app.k"), tmp_path) + assert not (tmp_path / "node_modules").exists() + + +def test_strict_raises_when_neither_exists(tmp_path: Path) -> None: + with pytest.raises(FileNotFoundError): + emit_frontend_types(_registry("app.k"), tmp_path, strict=True) + + +def test_published_target_merges_published_keys_with_app_keys(tmp_path: Path) -> None: + src = _published(tmp_path) + (src / "generated-resources.ts").write_text( + "export default {\n translation: {\n 'ui.save': '',\n 'users.title': '',\n" + " },\n} as const;\n" + ) + # App registry has its own namespace and no `users` — must not count as a removal. + emit_frontend_types(_registry("ui.save", "myapp.title"), tmp_path, strict=True) + resources = (src / "generated-resources.ts").read_text() + for key in ("ui.save", "users.title", "myapp.title"): + assert f"'{key}': ''" in resources + keys = (src / "keys.generated.ts").read_text() + assert "myapp" in keys and "users" in keys + + # Idempotent: a second run changes nothing. + mtime = (src / "keys.generated.ts").stat().st_mtime_ns + emit_frontend_types(_registry("ui.save", "myapp.title"), tmp_path, strict=True) + assert (src / "keys.generated.ts").stat().st_mtime_ns == mtime diff --git a/modules/audit_log/audit_log/models.py b/modules/audit_log/audit_log/models.py index 7230bfc9..d7f797cf 100644 --- a/modules/audit_log/audit_log/models.py +++ b/modules/audit_log/audit_log/models.py @@ -6,7 +6,8 @@ from datetime import UTC, datetime from simple_module_db.base import create_module_base -from sqlalchemy import JSON, DateTime, Index, func +from simple_module_db.types import AwareDateTime +from sqlalchemy import JSON, Index, func from sqlmodel import Column, Field from audit_log.constants import ( @@ -48,6 +49,6 @@ class AuditEntry(Base, table=True): # ty: ignore[unsupported-base] tenant_id: str | None = Field(default=None, max_length=TENANT_ID_MAX_LENGTH, index=True) created_at: datetime = Field( default_factory=lambda: datetime.now(UTC), - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), sa_column_kwargs={"server_default": func.now()}, ) diff --git a/modules/background_tasks/README.md b/modules/background_tasks/README.md index 2040e924..04f7eed6 100644 --- a/modules/background_tasks/README.md +++ b/modules/background_tasks/README.md @@ -15,6 +15,8 @@ Requires a Redis broker. The broker / result-backend URLs are module settings (` - Zero-config task discovery — any installed module that ships a `tasks.py` has its tasks autodiscovered (`celery.autodiscover_tasks` imports `.tasks` for every installed module). No per-module registration hook. - Admin UI at `/admin/background-tasks` — list recent runs, retry failed, inspect tracebacks (gated by the `background_tasks.view` permission). - `build_celery(settings)` factory in `background_tasks.celery_app`, plus `bind_task_context` / `get_log_context` / `install_log_filter` exported from the package root (import name `background_tasks`, distribution name `simple_module_background_tasks`). +- `enqueue_after_commit(db, task, *args, **kwargs)` — publish a task only once the session's transaction commits, never on rollback. +- `heartbeat()` — called from a long task body, keeps the stuck sweep from flagging it; `@shared_task(stuck_after=seconds)` sets a per-task stuck threshold. See the module docs for both. ## Usage diff --git a/modules/background_tasks/background_tasks/__init__.py b/modules/background_tasks/background_tasks/__init__.py index e7e2e9dd..49913f19 100644 --- a/modules/background_tasks/background_tasks/__init__.py +++ b/modules/background_tasks/background_tasks/__init__.py @@ -1,15 +1,19 @@ """BackgroundTasks module — Celery + Redis task queue with admin UI.""" +from background_tasks.after_commit import enqueue_after_commit from background_tasks.log_context import ( bind_task_context, get_log_context, install_log_filter, ) +from background_tasks.task_heartbeat import heartbeat from background_tasks.worker_settings import settings_for __all__ = [ "bind_task_context", + "enqueue_after_commit", "get_log_context", + "heartbeat", "install_log_filter", "settings_for", ] diff --git a/modules/background_tasks/background_tasks/after_commit.py b/modules/background_tasks/background_tasks/after_commit.py new file mode 100644 index 00000000..ad302817 --- /dev/null +++ b/modules/background_tasks/background_tasks/after_commit.py @@ -0,0 +1,122 @@ +"""Publish to the broker only once the transaction that justifies it commits. + +A task enqueued inside a transaction races that transaction: the worker can +pick the message up before the row it refers to is visible (eager mode runs +it *inside* ``send_task``), and if the commit then fails the task is already +queued against nothing. Both are fixed the same way — hold the publish until +the commit has landed, and drop it if the transaction rolls back (GH #452). + +Two delivery paths, picked per session: + +- **A request's session** (``get_db``) — the publish goes through + :meth:`~simple_module_db.RequestSession.on_commit`, which the framework runs + after the request's commit and before the response is sent, and discards on + rollback or commit failure. +- **Any other session** — a worker's sync session, a CLI script's own + ``async with session.begin()`` — has nobody to run ``on_commit`` callbacks, + so the publish hangs off SQLAlchemy's ``after_commit`` event instead and is + dropped when the outermost transaction ends without committing. + +The broker id is reserved up front and returned, so the caller can write it +onto its own row in the same transaction: the row names the task before the +task exists, never the other way round. +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid +from collections.abc import Callable, Sequence +from typing import Any + +from celery import current_app +from simple_module_db import RequestSession, bind_current_tenant +from simple_module_db.transaction import SESSION_START_KEY +from sqlalchemy import event +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Session, SessionTransaction + +logger = logging.getLogger(__name__) + +type Publish = Callable[[], object] + +_PENDING_KEY = "background_tasks_after_commit" +_HOOKED_KEY = "background_tasks_after_commit_hooked" + + +def enqueue_after_commit(db: AsyncSession | Session, task: Any, *args: Any, **kwargs: Any) -> str: + """Send *task* once *db*'s transaction commits; never if it rolls back. + + *task* is a Celery task (``send_receipt``), a signature + (``send_receipt.s().set(queue="mail")``, for routing options), or a + registered task name. ``*args`` / ``**kwargs`` are the task's own + arguments. Returns the Celery task id the message will carry. + + Publishes as the tenant bound *now*, so a task enqueued in a request runs + for that request's tenant even when the commit happens elsewhere. + """ + task_id = str(uuid.uuid4()) + + def publish() -> None: + if isinstance(task, str): + current_app.send_task(task, args=list(args), kwargs=kwargs, task_id=task_id) + else: + task.apply_async(args=args, kwargs=kwargs, task_id=task_id) + + after_commit(db, [bind_current_tenant(publish)]) + return task_id + + +def after_commit( + db: AsyncSession | Session, publishes: Sequence[Publish], *, offload: bool = False +) -> None: + """Run each of *publishes* after *db* commits, in order. Internal. + + *offload* sends the batch to a thread when the request path runs it: kombu + publishes block, and hundreds of them on the event loop stall every other + request. Left off for a single publish so Celery's eager mode — which runs + the task body inside the publish — stays on the loop's thread. One failed + publish is logged and does not stop the rest: the rows are already + committed, and an unsent ``pending`` row is visible and retryable. + """ + batch = list(publishes) + if isinstance(db, RequestSession) and SESSION_START_KEY in db.info: + if offload: + db.on_commit(lambda: asyncio.to_thread(_run_all, batch)) + else: + db.on_commit(lambda: _run_all(batch)) + return + sync = db.sync_session if isinstance(db, AsyncSession) else db + sync.info.setdefault(_PENDING_KEY, []).extend(batch) + _hook(sync) + + +def _run_all(batch: Sequence[Publish]) -> None: + for publish in batch: + try: + publish() + except Exception: + logger.exception("background_tasks.after_commit.publish_failed") + + +def _hook(session: Session) -> None: + """Attach the commit/rollback listeners to *session* once.""" + if session.info.get(_HOOKED_KEY): + return + session.info[_HOOKED_KEY] = True + + def on_commit(sess: Session) -> None: + _run_all(sess.info.pop(_PENDING_KEY, [])) + + def on_end(sess: Session, transaction: SessionTransaction) -> None: + # ``after_commit`` already took the batch on the commit path; anything + # still queued when the outermost transaction ends was rolled back. + if transaction.parent is None: + sess.info.pop(_PENDING_KEY, None) + + event.listen(session, "after_commit", on_commit) + event.listen(session, "after_transaction_end", on_end) + + +__all__ = ["enqueue_after_commit"] diff --git a/modules/background_tasks/background_tasks/constants.py b/modules/background_tasks/background_tasks/constants.py index df8799d6..1dae1556 100644 --- a/modules/background_tasks/background_tasks/constants.py +++ b/modules/background_tasks/background_tasks/constants.py @@ -84,6 +84,13 @@ class TaskStatus(StrEnum): DEFAULT_RESULT_BACKEND = "redis://localhost:6379/1" DEFAULT_STUCK_AFTER_SECONDS = 300 DEFAULT_STUCK_SWEEP_INTERVAL_SECONDS = 60 +# ``heartbeat()`` writes at most this often per task, however often a task body +# calls it. Keep stuck thresholds well above it. +HEARTBEAT_MIN_INTERVAL_SECONDS = 30 +# Task attribute giving one task its own stuck threshold, in seconds: +# ``@shared_task(name=..., stuck_after=4 * 3600)``. Celery copies extra +# decorator options onto the task class, so the sweep reads it off the registry. +TASK_STUCK_AFTER_ATTR = "stuck_after" DEFAULT_PURGE_INTERVAL_SECONDS = 60 * 60 * 24 DEFAULT_RETENTION_DAYS = 14 DEFAULT_MAX_RETRIES = 3 diff --git a/modules/background_tasks/background_tasks/endpoints/views.py b/modules/background_tasks/background_tasks/endpoints/views.py index 47e019f7..023da429 100644 --- a/modules/background_tasks/background_tasks/endpoints/views.py +++ b/modules/background_tasks/background_tasks/endpoints/views.py @@ -72,6 +72,12 @@ async def index( response.total == 0 and not task_name and not queue and not tenant_id and status is None ) + # `total` is the filtered count; the page must tell "the filters hide rows" + # from "nothing exists at all". + has_executions = response.total > 0 + if not has_executions and not unfiltered_and_empty: + has_executions = (await service.list(page=1, per_page=1)).total > 0 + async def strip_data() -> tuple[dict[str, int], list[str]]: """Everything above the table. Serial: one session, one connection.""" counts = await service.status_counts( @@ -108,6 +114,7 @@ async def strip_data() -> tuple[dict[str, int], list[str]]: "tenant_ids": await service.tenant_ids(), "platform_tenant_value": PLATFORM_TENANT_FILTER, "worker_presence": presence, + "has_executions": has_executions, "pagination": { "page": response.page, "per_page": response.per_page, diff --git a/modules/background_tasks/background_tasks/pages/Index.tsx b/modules/background_tasks/background_tasks/pages/Index.tsx index edfc47ac..66ec7f1a 100644 --- a/modules/background_tasks/background_tasks/pages/Index.tsx +++ b/modules/background_tasks/background_tasks/pages/Index.tsx @@ -41,6 +41,8 @@ interface Props { platform_tenant_value: string; /** Null unless the unfiltered list came back empty — see the index view. */ worker_presence: WorkerPresence | null; + /** Whether any execution exists at all, ignoring the active filters. */ + has_executions: boolean; } // A retry moves a row out of "failed", so the strip and the queue list are @@ -67,6 +69,7 @@ function Index() { tenant_ids: tenantIds, platform_tenant_value: platformTenantValue, worker_presence: workerPresence, + has_executions: hasExecutions, } = usePage<{ props: Props }>().props as unknown as Props; const { t } = useT(); @@ -222,7 +225,7 @@ function Index() { onPageChange={(page) => pushFilters(filters, page)} empty={ >(); + +/** + * Re-enqueue one execution. Clicks that land while a retry for the same row is + * still in flight share its request: the dialog's `busy` state only disables the + * button after a re-render, so a burst of clicks would otherwise each POST. + * Retrying a row again *after* the first retry settles stays allowed. + */ +export function retryExecution(execution: { id: string; task_name: string; }): Promise { + const pending = inFlight.get(execution.id); + if (pending) return pending; + const request = postRetry(execution).finally(() => inFlight.delete(execution.id)); + inFlight.set(execution.id, request); + return request; +} + +async function postRetry(execution: { id: string; task_name: string }): Promise { try { const res = await fetch(`${API_BASE}/executions/${execution.id}/retry`, { method: 'POST', diff --git a/modules/background_tasks/background_tasks/retry_service.py b/modules/background_tasks/background_tasks/retry_service.py index 96321f03..9599fa1c 100644 --- a/modules/background_tasks/background_tasks/retry_service.py +++ b/modules/background_tasks/background_tasks/retry_service.py @@ -8,15 +8,15 @@ from __future__ import annotations -import asyncio import uuid -from collections.abc import Sequence +from collections.abc import Callable, Sequence from typing import TYPE_CHECKING, Any from simple_module_core.events import EventBus from simple_module_db import RequestSession, all_tenants, tenant_context from sqlalchemy import Select, func, select +from background_tasks.after_commit import after_commit from background_tasks.constants import RETRY_ALL_BATCH, TaskStatus from background_tasks.contracts.events import TaskRetried from background_tasks.contracts.schemas import RetryFailedResult @@ -75,17 +75,25 @@ def __init__(self, db: RequestSession, celery: Celery, event_bus: EventBus) -> N self.event_bus = event_bus async def retry_one(self, row: TaskExecution) -> TaskExecution: - """Re-enqueue a single execution. - - Publishes inline rather than on a thread: it is one call, and keeping - it on the event loop keeps Celery's eager mode — which runs the task - body inside ``send_task`` — on the same thread the signal handlers are - bound to. The sweep below, which sends hundreds, goes to a thread. + """Re-enqueue a single execution, once the row recording it commits. + + The broker id is reserved so the row names the task before it is + sent; the publish itself waits for the commit (GH #452). Sending first + queued a task against a row that a failed commit then never wrote, + and a fast worker could start before the row was visible at all. + + Published inline from the post-commit hook rather than on a thread: + it is one call, and keeping it on the event loop keeps Celery's eager + mode — which runs the task body inside ``send_task`` — on the same + thread the signal handlers are bound to. The sweep below, which sends + hundreds, goes to a thread. """ - new_row = self._new_attempt(row, self._publish(row)) + task_id = str(uuid.uuid4()) + new_row = self._new_attempt(row, task_id) self.db.add(new_row) await self.db.flush() await self.db.refresh(new_row) + after_commit(self.db, [self._publisher(row, task_id)]) self._announce_after_commit([(row, new_row)]) return new_row @@ -114,9 +122,9 @@ async def retry_failed( Two operators pressing the button at once used to run every task twice: both read the same batch, and the child rows that would have excluded it were only written afterwards. :func:`claim_query` locks the batch, - and the children are written *before* the tasks are published — an + and the children are *committed* before the tasks are published — an orphaned ``pending`` row that was never sent is a far cheaper failure - than a task that ran twice. + than a task that ran twice, or one that ran with no row at all. """ # Read at call time, not bound as a default argument, so the cap has # exactly one source of truth that tests and operators can move. @@ -146,8 +154,7 @@ async def retry_failed( # The broker id is reserved here rather than read back from ``send_task`` # so the row recording the attempt can be written first and still name - # the task that will carry it. Held as plain strings, not read back off - # the ORM objects, because the publish below runs on another thread. + # the task that will carry it. reserved = [(row, str(uuid.uuid4())) for row in rows] new_rows = [self._new_attempt(row, task_id) for row, task_id in reserved] self.db.add_all(new_rows) @@ -155,43 +162,45 @@ async def retry_failed( # the sweep's real cost, and the rows have no reason to land singly. await self.db.flush() - # One thread for the whole batch. ``send_task`` is a blocking publish, - # and a few hundred of them inline would stall every other request for - # the length of the sweep. - await asyncio.to_thread( - lambda: [self._publish(row, task_id=task_id) for row, task_id in reserved] + # Sent only once the children commit (GH #452), and on one thread for + # the whole batch: ``send_task`` is a blocking publish, and a few + # hundred of them inline would stall every other request. + after_commit( + self.db, [self._publisher(row, task_id) for row, task_id in reserved], offload=True ) self._announce_after_commit(list(zip(rows, new_rows, strict=True))) return RetryFailedResult(queued=len(rows), remaining=max(0, eligible - len(rows))) - def _publish(self, row: TaskExecution, task_id: str | None = None) -> str: - """Send one task to the broker and return its Celery id. - - *task_id* pre-assigns that id, for callers that wrote the row naming it - before sending. Omitted, the broker assigns one and the caller stamps - the row with what comes back. + def _publisher(self, row: TaskExecution, task_id: str) -> Callable[[], str]: + """A call that sends *row*'s task again under the reserved *task_id*. - Blocking — kombu publishes synchronously. Call it off the event loop - when sending more than one. + Built while the row is attached and run after commit, possibly on + another thread — so everything it needs is read off the row now. + Blocking when run: kombu publishes synchronously. """ + name = row.task_name + tenant_id = row.tenant_id options: dict[str, Any] = { "args": list(row.args or []), "kwargs": dict(row.kwargs or {}), "queue": row.queue, + "task_id": task_id, } - if task_id is not None: - options["task_id"] = task_id - # Publish as the row's own tenant, never the operator's: a platform - # admin who is also a member of some org must not re-run another - # tenant's job stamped with theirs (``stamp_tenant`` would refuse the - # mismatch). A row with no tenant is a platform publish, so it goes out - # with none bound. - if row.tenant_id: - with tenant_context(row.tenant_id): - return self.celery.send_task(row.task_name, **options).id - with all_tenants(): - return self.celery.send_task(row.task_name, **options).id + + def publish() -> str: + # Publish as the row's own tenant, never the operator's: a platform + # admin who is also a member of some org must not re-run another + # tenant's job stamped with theirs (``stamp_tenant`` would refuse + # the mismatch). A row with no tenant is a platform publish, so it + # goes out with none bound. + if tenant_id: + with tenant_context(tenant_id): + return self.celery.send_task(name, **options).id + with all_tenants(): + return self.celery.send_task(name, **options).id + + return publish def _new_attempt(self, row: TaskExecution, celery_task_id: str) -> TaskExecution: """The row recording a fresh attempt at *row*. Not yet flushed.""" diff --git a/modules/background_tasks/background_tasks/signals.py b/modules/background_tasks/background_tasks/signals.py index bc1ae92f..0b433ce5 100644 --- a/modules/background_tasks/background_tasks/signals.py +++ b/modules/background_tasks/background_tasks/signals.py @@ -40,6 +40,7 @@ from background_tasks.log_context import signal_task_finished, signal_task_started from background_tasks.models import TaskExecution from background_tasks.sync_db import sync_session +from background_tasks.task_heartbeat import forget_heartbeat from background_tasks.tenant_context import ( message_tenant, published_tenant, @@ -172,6 +173,7 @@ def on_task_postrun( tenant_id=message_tenant(task=task), ) release_tenant(task_id=task_id) + forget_heartbeat(task_id) signal_task_finished(task_id=task_id) diff --git a/modules/background_tasks/background_tasks/task_heartbeat.py b/modules/background_tasks/background_tasks/task_heartbeat.py new file mode 100644 index 00000000..158d087c --- /dev/null +++ b/modules/background_tasks/background_tasks/task_heartbeat.py @@ -0,0 +1,98 @@ +"""Liveness for long-running task bodies (GH #451). + +The signal handlers stamp ``heartbeat_at`` when a task starts and when it +finishes — nothing in between. :func:`~background_tasks.tasks.sweep_stuck_tasks` +treats a ``running`` row whose heartbeat is older than the stuck threshold as +a worker that died, and makes it retry-eligible. Without a way to check in +from *inside* the task, every job longer than the threshold was flagged stuck +and could be run a second time while the first was still working. + +A task body calls :func:`heartbeat` from its loop. Writes are throttled per +task, in-process, so calling it on every iteration is cheap; and it is a +no-op outside a running task, so shared helpers can call it unconditionally. +""" + +from __future__ import annotations + +import logging +import threading +import time + +from celery import current_task +from sqlalchemy import update + +from background_tasks._signal_support import now_utc +from background_tasks.constants import HEARTBEAT_MIN_INTERVAL_SECONDS, TaskStatus +from background_tasks.models import TaskExecution +from background_tasks.sync_db import sync_session + +logger = logging.getLogger(__name__) + +_last_beat: dict[str, float] = {} +_lock = threading.Lock() + + +def _current_task_id() -> str | None: + """The id of the Celery task executing on this thread, if any. + + ``called_directly`` is set when a task function is invoked as a plain + call (``my_task()``) — there is no execution row to refresh then. + """ + try: + request = current_task.request + except AttributeError: # no task bound: the proxy wraps ``None`` + return None + if request is None or getattr(request, "called_directly", True): + return None + task_id = getattr(request, "id", None) + return task_id if isinstance(task_id, str) and task_id else None + + +def heartbeat(*, force: bool = False) -> bool: + """Tell the stuck sweep this task is still alive. + + Refreshes ``heartbeat_at`` on the current task's execution row, at most + once per :data:`~background_tasks.constants.HEARTBEAT_MIN_INTERVAL_SECONDS` + per task unless *force* is set. Call it more often than the task's stuck + threshold — every loop iteration is fine. + + Returns whether a write was made. Never raises: outside a task it does + nothing, and a failed write is logged rather than failing the job it is + reporting on. + """ + task_id = _current_task_id() + if task_id is None: + return False + now = time.monotonic() + with _lock: + last = _last_beat.get(task_id) + if not force and last is not None and now - last < HEARTBEAT_MIN_INTERVAL_SECONDS: + return False + _last_beat[task_id] = now + try: + with sync_session() as session: + session.execute( + update(TaskExecution) + .where( + TaskExecution.celery_task_id == task_id, + # Only a row still ``running``: a heartbeat must not revive + # a row the sweep already declared stuck (it may have been + # retried since) or rewrite a terminal one. + TaskExecution.status == TaskStatus.RUNNING, + ) + .values(heartbeat_at=now_utc()) + ) + except Exception: + logger.exception("heartbeat failed for task_id=%s", task_id) + return False + return True + + +def forget_heartbeat(task_id: str | None) -> None: + """Drop *task_id*'s throttle entry once the task has finished.""" + if task_id: + with _lock: + _last_beat.pop(task_id, None) + + +__all__ = ["forget_heartbeat", "heartbeat"] diff --git a/modules/background_tasks/background_tasks/tasks.py b/modules/background_tasks/background_tasks/tasks.py index 1f12299f..8f4f3342 100644 --- a/modules/background_tasks/background_tasks/tasks.py +++ b/modules/background_tasks/background_tasks/tasks.py @@ -12,10 +12,13 @@ import logging import os +from collections import defaultdict +from collections.abc import Mapping from datetime import UTC, datetime, timedelta +from typing import Any -from celery import shared_task -from sqlalchemy import delete, update +from celery import Task, shared_task +from sqlalchemy import ColumnElement, and_, delete, or_, update from background_tasks.constants import ( DEFAULT_RETENTION_DAYS, @@ -23,6 +26,7 @@ DEMO_ECHO_TASK, INTERNAL_TASK_PURGE_OLD, INTERNAL_TASK_SWEEP_STUCK, + TASK_STUCK_AFTER_ATTR, TERMINAL_STATUSES, TaskStatus, ) @@ -54,10 +58,52 @@ def demo_echo(message: str = "hello") -> dict[str, str]: return {"message": message, "at": datetime.now(UTC).isoformat()} -@shared_task(name=INTERNAL_TASK_SWEEP_STUCK) -def sweep_stuck_tasks() -> int: +def stuck_thresholds(registry: Mapping[str, Any]) -> dict[str, int]: + """Per-task stuck thresholds declared on the tasks in *registry*. + + A task opts in with ``@shared_task(name=..., stuck_after=seconds)``; Celery + copies the option onto the task class. Read from the registry of the + process running the sweep, which autodiscovers every module's tasks, so + the threshold lives next to the task and needs no column or migration. + """ + found: dict[str, int] = {} + for name, task in registry.items(): + value = getattr(task, TASK_STUCK_AFTER_ATTR, None) + if isinstance(value, int | float) and not isinstance(value, bool) and value > 0: + found[name] = int(value) + return found + + +def _stale(now: datetime, stuck_after: int, overrides: dict[str, int]) -> ColumnElement[bool]: + """Rows whose heartbeat is older than their own task's threshold.""" + clauses = [ + and_( + TaskExecution.task_name.in_(names), + TaskExecution.heartbeat_at < now - timedelta(seconds=seconds), + ) + for seconds, names in _group_by_threshold(overrides).items() + ] + default = TaskExecution.heartbeat_at < now - timedelta(seconds=stuck_after) + if overrides: + default = and_(TaskExecution.task_name.not_in(list(overrides)), default) + return or_(default, *clauses) + + +def _group_by_threshold(overrides: dict[str, int]) -> dict[int, list[str]]: + groups: defaultdict[int, list[str]] = defaultdict(list) + for name, seconds in overrides.items(): + groups[seconds].append(name) + return groups + + +@shared_task(name=INTERNAL_TASK_SWEEP_STUCK, bind=True) +def sweep_stuck_tasks(self: Task) -> int: """Flip ``running`` rows with stale heartbeats to ``stuck``. + "Stale" is per task: a task's own ``stuck_after`` (see + :func:`stuck_thresholds`) wins over the global setting. Long jobs should + also call :func:`background_tasks.heartbeat` as they go. + Returns the number of rows flipped — handy for metrics/log scraping. """ # Avoid import-time coupling to settings (beat schedules this task @@ -65,16 +111,17 @@ def sweep_stuck_tasks() -> int: stuck_after = int( os.environ.get("SM_BG_TASKS_STUCK_AFTER_SECONDS", DEFAULT_STUCK_AFTER_SECONDS) ) - cutoff = datetime.now(UTC) - timedelta(seconds=stuck_after) + now = datetime.now(UTC) + overrides = stuck_thresholds(self.app.tasks) with sync_session() as session: stmt = ( update(TaskExecution) .where( TaskExecution.status == TaskStatus.RUNNING, - TaskExecution.heartbeat_at < cutoff, + _stale(now, stuck_after, overrides), ) - .values(status=TaskStatus.STUCK, finished_at=datetime.now(UTC)) + .values(status=TaskStatus.STUCK, finished_at=now) ) result = session.execute(stmt) count = result.rowcount or 0 diff --git a/modules/background_tasks/tests-js/retry.test.ts b/modules/background_tasks/tests-js/retry.test.ts new file mode 100644 index 00000000..a421bab3 --- /dev/null +++ b/modules/background_tasks/tests-js/retry.test.ts @@ -0,0 +1,31 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('sonner', () => ({ toast: { success: vi.fn(), error: vi.fn() } })); + +import { retryExecution } from '../background_tasks/pages/retry'; + +const row = { id: 'abc', task_name: 'demo.task' }; + +function stubFetch() { + const fetchMock = vi.fn(async () => new Response(JSON.stringify({ id: 'new' }), { status: 200 })); + vi.stubGlobal('fetch', fetchMock); + return fetchMock; +} + +describe('retryExecution', () => { + afterEach(() => vi.unstubAllGlobals()); + + it('sends one POST for a burst of clicks on the same row', async () => { + const fetchMock = stubFetch(); + const results = await Promise.all([1, 2, 3, 4, 5].map(() => retryExecution(row))); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(results.every((r) => r?.id === 'new')).toBe(true); + }); + + it('allows retrying the same row again once the first request settled', async () => { + const fetchMock = stubFetch(); + await retryExecution(row); + await retryExecution(row); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); +}); diff --git a/modules/background_tasks/tests/test_background_tasks_after_commit.py b/modules/background_tasks/tests/test_background_tasks_after_commit.py new file mode 100644 index 00000000..ee9dde3b --- /dev/null +++ b/modules/background_tasks/tests/test_background_tasks_after_commit.py @@ -0,0 +1,216 @@ +"""Nothing reaches the broker before the transaction that justifies it (GH #452). + +Retrying used to ``send_task`` while the new ``TaskExecution`` row was only +flushed: a failed commit left a task queued against no row, and a fast worker +could start before the row was visible. These pin the rule for both retry +paths and for :func:`background_tasks.enqueue_after_commit`, the helper that +gives app code the same guarantee. +""" + +from __future__ import annotations + +import time +import uuid +from typing import Any +from unittest.mock import MagicMock + +import httpx +import pytest +from background_tasks import enqueue_after_commit, sync_db +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution +from background_tasks.service import BackgroundTaskService +from simple_module_core.events import EventBus +from simple_module_db import RequestSession, finalize_session +from simple_module_db.transaction import SESSION_START_KEY, rollback_session +from sqlalchemy import event + +ADMIN_BASE = "/api/background_tasks/admin" + + +@pytest.fixture(params=["request-managed", "unmanaged"]) +def session(request: pytest.FixtureRequest, db_session: RequestSession) -> RequestSession: + """The same session, once as ``get_db`` hands it out and once bare. + + ``get_db`` stamps :data:`SESSION_START_KEY`; that is what routes the + publish through ``on_commit``. A bare session falls back to SQLAlchemy's + own commit events — both must hold the publish until the commit. + """ + if request.param == "request-managed": + db_session.info[SESSION_START_KEY] = time.perf_counter() + return db_session + + +@pytest.fixture +def celery() -> MagicMock: + celery = MagicMock(name="Celery") + celery.send_task.side_effect = lambda name, **options: MagicMock(id=options.get("task_id")) + return celery + + +async def _failed_row(db: RequestSession) -> TaskExecution: + row = TaskExecution( + celery_task_id=str(uuid.uuid4()), + task_name="orders.ship", + status=TaskStatus.FAILED, + args=[1], + kwargs={}, + ) + db.add(row) + await db.flush() + await db.refresh(row) + return row + + +class TestRetryOne: + async def test_published_once_after_commit( + self, session: RequestSession, celery: MagicMock + ) -> None: + service = BackgroundTaskService(db=session, celery=celery, event_bus=EventBus()) + original = await _failed_row(session) + + detail = await service.retry(original.id) + assert celery.send_task.call_count == 0 + + await finalize_session(session) + + assert celery.send_task.call_count == 1 + assert celery.send_task.call_args.kwargs["task_id"] == detail.celery_task_id + + async def test_nothing_published_on_rollback( + self, session: RequestSession, celery: MagicMock + ) -> None: + service = BackgroundTaskService(db=session, celery=celery, event_bus=EventBus()) + original = await _failed_row(session) + + await service.retry(original.id) + await rollback_session(session) + # A later, unrelated commit on the same session must not resurrect it. + await session.commit() + + celery.send_task.assert_not_called() + + async def test_nothing_published_when_the_commit_fails( + self, session: RequestSession, celery: MagicMock + ) -> None: + service = BackgroundTaskService(db=session, celery=celery, event_bus=EventBus()) + original = await _failed_row(session) + await service.retry(original.id) + + def _refuse(*_a: Any) -> None: + raise RuntimeError("commit refused") + + sync_engine = session.bind.sync_engine + event.listen(sync_engine, "commit", _refuse) + try: + with pytest.raises(RuntimeError, match="commit refused"): + await finalize_session(session) + finally: + event.remove(sync_engine, "commit", _refuse) + + celery.send_task.assert_not_called() + + +@pytest.mark.usefixtures("_stub_celery") +class TestThroughTheApi: + """The real request path: ``get_db`` + ``CommitBeforeResponseMiddleware``.""" + + @pytest.fixture + def trace(self, app) -> list[str]: + trace: list[str] = [] + engine = app.state.sm.db.engine.sync_engine + + def _send_task(name: str, **options: Any) -> MagicMock: + trace.append("publish") + return MagicMock(id=options.get("task_id")) + + app.state.background_tasks.celery.send_task.side_effect = _send_task + event.listen(engine, "commit", lambda _conn: trace.append("commit")) + return trace + + async def test_single_retry_publishes_once_after_the_commit( + self, trace, seed_execution, execution_rows, authenticated_client: httpx.AsyncClient + ) -> None: + original = await seed_execution(status=TaskStatus.FAILED) + trace.clear() + + resp = await authenticated_client.post(f"{ADMIN_BASE}/executions/{original.id}/retry") + + assert resp.status_code == 200 + assert trace.count("publish") == 1 + assert trace.index("commit") < trace.index("publish") + new_row = next(r for r in await execution_rows() if r.id != original.id) + assert new_row.celery_task_id == resp.json()["celery_task_id"] + + async def test_bulk_retry_publishes_after_the_commit( + self, trace, seed_execution, authenticated_client: httpx.AsyncClient + ) -> None: + for _ in range(3): + await seed_execution(status=TaskStatus.FAILED) + trace.clear() + + resp = await authenticated_client.post(f"{ADMIN_BASE}/executions/retry-failed") + + assert resp.json() == {"queued": 3, "remaining": 0} + assert trace.count("publish") == 3 + assert trace.index("commit") < trace.index("publish") + + +class TestEnqueueAfterCommit: + async def test_request_session_publishes_after_commit(self, session: RequestSession) -> None: + task = MagicMock(name="task") + + task_id = enqueue_after_commit(session, task, 42, priority="high") + session.add(TaskExecution(celery_task_id=task_id, task_name="orders.ship")) + await session.flush() + task.apply_async.assert_not_called() + + await finalize_session(session) + + task.apply_async.assert_called_once_with( + args=(42,), kwargs={"priority": "high"}, task_id=task_id + ) + + async def test_request_session_rollback_drops_it(self, session: RequestSession) -> None: + task = MagicMock(name="task") + enqueue_after_commit(session, task) + session.add(TaskExecution(task_name="orders.ship")) + await session.flush() + + await rollback_session(session) + + task.apply_async.assert_not_called() + + def test_worker_sync_session(self, sync_sqlite) -> None: + """A task body enqueuing follow-up work from its own sync session.""" + task = MagicMock(name="task") + factory = sync_db.get_sync_session_factory() + + with factory() as session: + enqueue_after_commit(session, task, 1) + session.add(TaskExecution(task_name="orders.ship")) + session.flush() + task.apply_async.assert_not_called() + session.commit() + task.apply_async.assert_called_once() + + with factory() as session: + enqueue_after_commit(session, task, 2) + session.add(TaskExecution(task_name="orders.ship")) + session.flush() + session.rollback() + session.commit() + assert task.apply_async.call_count == 1 + + async def test_a_task_name_is_sent_through_the_current_app( + self, session: RequestSession, monkeypatch: pytest.MonkeyPatch + ) -> None: + app = MagicMock(name="Celery") + monkeypatch.setattr("background_tasks.after_commit.current_app", app) + + task_id = enqueue_after_commit(session, "orders.ship", 7) + session.add(TaskExecution(task_name="orders.ship")) + await session.flush() + await finalize_session(session) + + app.send_task.assert_called_once_with("orders.ship", args=[7], kwargs={}, task_id=task_id) diff --git a/modules/background_tasks/tests/test_background_tasks_heartbeat.py b/modules/background_tasks/tests/test_background_tasks_heartbeat.py new file mode 100644 index 00000000..a5c340be --- /dev/null +++ b/modules/background_tasks/tests/test_background_tasks_heartbeat.py @@ -0,0 +1,152 @@ +"""Long jobs stay ``running`` while they say so (GH #451). + +``heartbeat_at`` used to be stamped only at start and end, so every task +longer than the stuck threshold was flipped to ``stuck`` — retry-eligible +while it was still working. A task body now checks in with +:func:`background_tasks.heartbeat`, and a task can declare its own threshold +with ``@shared_task(stuck_after=...)``. +""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime, timedelta + +import pytest +from background_tasks import heartbeat, sync_db +from background_tasks import task_heartbeat as heartbeat_mod +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution +from background_tasks.tasks import stuck_thresholds, sweep_stuck_tasks +from celery import Celery, shared_task +from sqlalchemy import select + +LONG_JOB = "test_bg_heartbeat.long_job" +TOUCHY_JOB = "test_bg_heartbeat.touchy_job" + + +@shared_task(name=LONG_JOB, stuck_after=3600) +def long_job() -> None: ... + + +@shared_task(name=TOUCHY_JOB, stuck_after=10) +def touchy_job() -> None: ... + + +@pytest.fixture(autouse=True) +def _global_threshold_60s(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SM_BG_TASKS_STUCK_AFTER_SECONDS", "60") + + +@pytest.fixture(autouse=True) +def _fresh_throttle() -> None: + heartbeat_mod._last_beat.clear() + + +def _running_row(*, task_name: str = "test.task", heartbeat_age: float) -> str: + """Insert a ``running`` row last heard from *heartbeat_age* seconds ago.""" + now = datetime.now(UTC) + celery_id = str(uuid.uuid4()) + with sync_db.get_sync_session_factory()() as session: + session.add( + TaskExecution( + celery_task_id=celery_id, + task_name=task_name, + status=TaskStatus.RUNNING, + started_at=now - timedelta(hours=2), + heartbeat_at=now - timedelta(seconds=heartbeat_age), + ) + ) + session.commit() + return celery_id + + +def _status(celery_id: str) -> TaskStatus: + with sync_db.get_sync_session_factory()() as session: + stmt = select(TaskExecution.status).where(TaskExecution.celery_task_id == celery_id) + return session.execute(stmt).scalar_one() + + +def _inside(monkeypatch: pytest.MonkeyPatch, celery_id: str) -> None: + """Pretend the current thread is executing *celery_id*.""" + monkeypatch.setattr(heartbeat_mod, "_current_task_id", lambda: celery_id) + + +class TestHeartbeat: + def test_is_a_no_op_outside_a_task(self, sync_sqlite) -> None: + assert heartbeat() is False + + def test_a_task_that_heartbeats_past_the_threshold_is_not_stuck( + self, sync_sqlite, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Started two hours ago against a 60 s threshold — alive because it said so.""" + celery_id = _running_row(heartbeat_age=120) + _inside(monkeypatch, celery_id) + + assert heartbeat() is True + assert sweep_stuck_tasks() == 0 + assert _status(celery_id) == TaskStatus.RUNNING + + def test_a_silent_task_past_the_threshold_is_stuck(self, sync_sqlite) -> None: + celery_id = _running_row(heartbeat_age=120) + + assert sweep_stuck_tasks() == 1 + assert _status(celery_id) == TaskStatus.STUCK + + def test_writes_are_throttled_per_task( + self, sync_sqlite, monkeypatch: pytest.MonkeyPatch + ) -> None: + celery_id = _running_row(heartbeat_age=0) + _inside(monkeypatch, celery_id) + + assert heartbeat() is True + assert heartbeat() is False + assert heartbeat(force=True) is True + + def test_does_not_revive_a_row_already_declared_stuck( + self, sync_sqlite, monkeypatch: pytest.MonkeyPatch + ) -> None: + celery_id = _running_row(heartbeat_age=120) + sweep_stuck_tasks() + _inside(monkeypatch, celery_id) + + heartbeat() + + assert _status(celery_id) == TaskStatus.STUCK + + def test_finds_the_running_task_by_itself(self, sync_sqlite) -> None: + """Through a real task execution, not a patched id lookup.""" + app = Celery("heartbeat-test", set_as_current=False) + seen: list[str | None] = [] + + @app.task(name="test_bg_heartbeat.beating", bind=True) + def beating(self) -> None: + seen.append(heartbeat_mod._current_task_id()) + seen.append(self.request.id) + + beating.apply(task_id="beat-1") + + assert seen == ["beat-1", "beat-1"] + assert heartbeat_mod._current_task_id() is None + + +class TestPerTaskThreshold: + def test_reads_stuck_after_off_the_registry(self) -> None: + registry = {LONG_JOB: long_job, TOUCHY_JOB: touchy_job, "x": object()} + assert stuck_thresholds(registry) == {LONG_JOB: 3600, TOUCHY_JOB: 10} + + def test_a_longer_threshold_overrides_the_global_one(self, sync_sqlite) -> None: + within = _running_row(task_name=LONG_JOB, heartbeat_age=120) + beyond = _running_row(task_name=LONG_JOB, heartbeat_age=2 * 3600) + + assert sweep_stuck_tasks() == 1 + assert _status(within) == TaskStatus.RUNNING + assert _status(beyond) == TaskStatus.STUCK + + def test_a_shorter_threshold_overrides_the_global_one(self, sync_sqlite) -> None: + touchy = _running_row(task_name=TOUCHY_JOB, heartbeat_age=30) + ordinary = _running_row(heartbeat_age=30) + + assert sweep_stuck_tasks() == 1 + assert _status(touchy) == TaskStatus.STUCK + assert _status(ordinary) == TaskStatus.RUNNING diff --git a/modules/background_tasks/tests/test_bg_index_has_executions.py b/modules/background_tasks/tests/test_bg_index_has_executions.py new file mode 100644 index 00000000..095f2d1d --- /dev/null +++ b/modules/background_tasks/tests/test_bg_index_has_executions.py @@ -0,0 +1,62 @@ +"""The index view's ``has_executions`` prop (GH #469). + +``pagination.total`` is the *filtered* count, so a search on an install with no +executions at all looked identical to a search hiding existing rows, and the +empty state claimed rows "exist outside the current search". The prop carries +the unfiltered answer. +""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime + +import httpx +import pytest +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution + +VIEW_BASE = "/admin/background-tasks/" +INERTIA_HEADERS = {"X-Inertia": "true", "Accept": "application/json"} + +pytestmark = pytest.mark.usefixtures("_stub_celery") + + +async def _props(client: httpx.AsyncClient, **params: str) -> dict: + resp = await client.get(VIEW_BASE, params=params, headers=INERTIA_HEADERS) + assert resp.status_code == 200 + return resp.json()["props"] + + +async def _seed(app) -> None: + async with app.state.sm.db.session_factory() as session: + session.add( + TaskExecution( + celery_task_id=str(uuid.uuid4()), + task_name="demo.a", + status=TaskStatus.FAILED, + queue="default", + args=[], + kwargs={}, + queued_at=datetime(2026, 5, 1, 12, 0, tzinfo=UTC), + ) + ) + await session.commit() + + +async def test_empty_install_reports_none_even_when_searching( + authenticated_client: httpx.AsyncClient, +) -> None: + assert (await _props(authenticated_client))["has_executions"] is False + assert (await _props(authenticated_client, q="anything"))["has_executions"] is False + + +async def test_filters_hiding_existing_rows_still_report_some( + app, authenticated_client: httpx.AsyncClient +) -> None: + await _seed(app) + + hidden = await _props(authenticated_client, q="no-such-task") + assert hidden["pagination"]["total"] == 0 + assert hidden["has_executions"] is True + assert (await _props(authenticated_client))["has_executions"] is True diff --git a/modules/background_tasks/tests/test_bg_service.py b/modules/background_tasks/tests/test_bg_service.py index 3ec6b69d..f0a31beb 100644 --- a/modules/background_tasks/tests/test_bg_service.py +++ b/modules/background_tasks/tests/test_bg_service.py @@ -185,21 +185,23 @@ async def _on_retried(event: TaskRetried) -> None: detail = await service.retry(original.id) + assert detail.retried_from_id == original.id + assert detail.status == TaskStatus.PENDING + # Nothing is sent or announced while the retry is still a pending + # write — a worker or subscriber acting on a transaction that rolls + # back acts on a retry that never happened (GH #452). + mock_celery.send_task.assert_not_called() + assert received == [] + + await finalize_session(db_session) + mock_celery.send_task.assert_called_once_with( "orders.ship", args=[42], kwargs={"priority": "high"}, queue="orders", + task_id=detail.celery_task_id, ) - assert detail.retried_from_id == original.id - assert detail.status == TaskStatus.PENDING - assert detail.celery_task_id == "new-celery-id-123" - # Nothing is announced while the retry is still a pending write — a - # subscriber acting on a transaction that rolls back acts on a retry - # that never happened. - assert received == [] - - await finalize_session(db_session) assert len(received) == 1 assert received[0].original_id == original.id diff --git a/modules/background_tasks/tests/test_execution_tenancy.py b/modules/background_tasks/tests/test_execution_tenancy.py index df527138..6338baec 100644 --- a/modules/background_tasks/tests/test_execution_tenancy.py +++ b/modules/background_tasks/tests/test_execution_tenancy.py @@ -85,13 +85,13 @@ class TestRetryPublish: def test_publishes_as_the_row_tenant_not_the_operators(self): coordinator, seen = _coordinator() with tenant_context("operators-org"): - coordinator._publish(_row("acme")) + coordinator._publisher(_row("acme"), "new-id")() assert seen == ["acme"] def test_a_platform_row_is_published_with_no_tenant(self): coordinator, seen = _coordinator() with tenant_context("operators-org"): - coordinator._publish(_row(None)) + coordinator._publisher(_row(None), "new-id")() assert seen == [None] def test_the_new_attempt_copies_the_tenant(self): diff --git a/modules/file_storage/file_storage/backends/filesystem.py b/modules/file_storage/file_storage/backends/filesystem.py index b21c3665..a5e74c1e 100644 --- a/modules/file_storage/file_storage/backends/filesystem.py +++ b/modules/file_storage/file_storage/backends/filesystem.py @@ -65,11 +65,17 @@ async def put( tmp.unlink(missing_ok=True) raise - async def get(self, key: str) -> AsyncIterator[bytes]: + def _existing(self, key: str) -> Path: path = self._resolve(key) if not path.exists(): raise StorageNotFoundError(key) - return _stream_file(path) + return path + + async def get(self, key: str) -> AsyncIterator[bytes]: + return _stream_file(self._existing(key)) + + async def get_range(self, key: str, start: int, end: int) -> AsyncIterator[bytes]: + return _stream_file(self._existing(key), start=start, length=end - start + 1) async def delete(self, key: str) -> None: path = self._resolve(key) @@ -82,12 +88,22 @@ async def presigned_get_url(self, key: str, ttl_seconds: int) -> str: raise NotSupportedError("Filesystem backend cannot mint presigned URLs.") -async def _stream_file(path: Path) -> AsyncIterator[bytes]: +async def _stream_file( + path: Path, *, start: int = 0, length: int | None = None +) -> AsyncIterator[bytes]: + """Yield ``length`` bytes from ``start`` (to EOF when ``None``), chunk by chunk.""" + remaining = length async with aiofiles.open(path, "rb") as fh: - while True: - chunk = await fh.read(constants.DEFAULT_CHUNK_SIZE) + await fh.seek(start) + while remaining is None or remaining > 0: + size = constants.DEFAULT_CHUNK_SIZE + if remaining is not None: + size = min(size, remaining) + chunk = await fh.read(size) if not chunk: break + if remaining is not None: + remaining -= len(chunk) yield chunk diff --git a/modules/file_storage/file_storage/backends/s3.py b/modules/file_storage/file_storage/backends/s3.py index 95273c7d..1ad5cbe2 100644 --- a/modules/file_storage/file_storage/backends/s3.py +++ b/modules/file_storage/file_storage/backends/s3.py @@ -7,22 +7,20 @@ from __future__ import annotations -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator, AsyncIterator from tempfile import SpooledTemporaryFile -from typing import TYPE_CHECKING, Any +from typing import Any from file_storage import constants from file_storage.backends import register_backend from file_storage.contracts.service import ( ConfigurationError, StorageBackendError, + StorageError, StorageNotFoundError, ) from file_storage.settings import FileStorageSettings -if TYPE_CHECKING: - pass - _MISSING_S3_EXTRA = ( "S3 backend requires the 'aioboto3' package. Install with: uv add --optional s3 aioboto3" ) @@ -83,20 +81,37 @@ async def put( raise StorageBackendError(f"S3 put_object failed: {exc}") from exc async def get(self, key: str) -> AsyncIterator[bytes]: + return await _open(self._stream(key, {})) + + async def get_range(self, key: str, start: int, end: int) -> AsyncIterator[bytes]: + return await _open(self._stream(key, {"Range": f"bytes={start}-{end}"})) + + async def _stream(self, key: str, extra: dict[str, str]) -> AsyncGenerator[bytes]: + """Stream ``get_object``'s body in ``DEFAULT_CHUNK_SIZE`` reads. + + Never ``Body.read()`` without a size: that buffers the whole object, + and these can be multi-GB rasters. The client and the HTTP body stay + open only while the generator runs; closing it — exhaustion, an + error, or the server cancelling on client disconnect — releases both. + """ try: async with self._client() as client: - resp = await client.get_object(Bucket=self.bucket, Key=key) - # ``Body.read()`` works uniformly across aioboto3 (real) and - # moto's patched aiobotocore (tests); ``iter_chunks`` is sync - # under moto, breaking ``async for``. - data = await resp["Body"].read() + try: + resp = await client.get_object(Bucket=self.bucket, Key=key, **extra) + except Exception as exc: + if _is_not_found(exc): + raise StorageNotFoundError(key) from exc + raise StorageBackendError(f"S3 get_object failed: {exc}") from exc + body = resp["Body"] + try: + while chunk := await body.read(constants.DEFAULT_CHUNK_SIZE): + yield chunk + finally: + body.close() + except StorageError: + raise except Exception as exc: - if _is_not_found(exc): - raise StorageNotFoundError(key) from exc - raise StorageBackendError(f"S3 get_object failed: {exc}") from exc - - for offset in range(0, len(data), constants.DEFAULT_CHUNK_SIZE): - yield data[offset : offset + constants.DEFAULT_CHUNK_SIZE] + raise StorageBackendError(f"S3 get_object stream failed: {exc}") from exc async def delete(self, key: str) -> None: try: @@ -127,6 +142,31 @@ async def presigned_get_url(self, key: str, ttl_seconds: int) -> str: raise StorageBackendError(f"presign failed: {exc}") from exc +async def _open(stream: AsyncGenerator[bytes]) -> AsyncIterator[bytes]: + """Start ``stream`` now so a missing key raises before any response begins. + + Pulling the first chunk sends ``get_object``; a ``StorageNotFoundError`` + therefore surfaces from the awaited ``get``/``get_range`` call, where the + endpoints turn it into a 404, instead of after a 200 is already on the wire. + """ + iterator = aiter(stream) + try: + first = await anext(iterator) + except StopAsyncIteration: + return _chain(None, iterator) + return _chain(first, iterator) + + +async def _chain(first: bytes | None, rest: AsyncGenerator[bytes]) -> AsyncIterator[bytes]: + try: + if first is not None: + yield first + async for chunk in rest: + yield chunk + finally: + await rest.aclose() + + def _is_not_found(exc: Exception) -> bool: """Detect a 404 from botocore's heterogeneous error shapes.""" response = getattr(exc, "response", None) diff --git a/modules/file_storage/file_storage/contracts/service.py b/modules/file_storage/file_storage/contracts/service.py index ada82c57..e818efc8 100644 --- a/modules/file_storage/file_storage/contracts/service.py +++ b/modules/file_storage/file_storage/contracts/service.py @@ -5,8 +5,8 @@ means writing one module that implements this Protocol and self-registers via :func:`file_storage.backends.register_backend`. -The Protocol is deliberately narrow: ``put``, ``get``, ``delete``, ``exists``, -plus capability flags so the service layer can dispatch (e.g. presigned-URL +The Protocol is deliberately narrow: ``put``, ``get``, ``get_range``, +``delete``, ``exists``, plus capability flags so the service layer can dispatch (e.g. presigned-URL download vs proxied stream) without inspecting backend identity. """ @@ -65,7 +65,22 @@ async def put( """Persist ``stream`` under ``key``. ``size`` is the total byte count.""" async def get(self, key: str) -> AsyncIterator[bytes]: - """Yield the object's bytes in chunks. Raises :class:`StorageNotFoundError`.""" + """Return an iterator over the object's bytes in bounded chunks. + + Awaiting the call opens the object, so a missing key raises + :class:`StorageNotFoundError` here rather than mid-stream. The object + must be streamed, never buffered whole: files can be many GB. + """ + + async def get_range(self, key: str, start: int, end: int) -> AsyncIterator[bytes]: + """Like :meth:`get`, for bytes ``start`` through ``end`` *inclusive*. + + HTTP ``Range`` semantics: ``get_range(k, 0, 0)`` is the first byte. The + caller has already validated ``0 <= start <= end < size``; backends need + not re-check against the object's size. Optional for third-party + backends — without it the download endpoints ignore ``Range`` and send + the whole object. + """ async def delete(self, key: str) -> None: """Remove the object at ``key``. No-op if absent.""" diff --git a/modules/file_storage/file_storage/endpoints/api.py b/modules/file_storage/file_storage/endpoints/api.py index 3fe3c824..f599537f 100644 --- a/modules/file_storage/file_storage/endpoints/api.py +++ b/modules/file_storage/file_storage/endpoints/api.py @@ -12,11 +12,12 @@ Form, HTTPException, Query, + Request, Response, UploadFile, status, ) -from fastapi.responses import RedirectResponse, StreamingResponse +from fastapi.responses import RedirectResponse from simple_module_core.events import EventBus from simple_module_hosting.i18n_deps import TranslatorDep from simple_module_hosting.permissions import RequiresPermission @@ -30,8 +31,10 @@ StoredFileOut, StoredFileUpdate, ) +from file_storage.contracts.service import StorageNotFoundError from file_storage.deps import get_event_bus, get_file_storage_service from file_storage.format import format_bytes +from file_storage.ranges import file_body_response from file_storage.service import ( ContentTypeNotAllowedError, FileStorageService, @@ -40,7 +43,7 @@ StoredFileNotFoundError, StreamDownload, ) -from file_storage.serving import not_found, thumbnail_response +from file_storage.serving import etag_for, not_found, thumbnail_response router = APIRouter() @@ -188,28 +191,30 @@ async def file_thumbnail( ) async def download_file( file_id: uuid.UUID, + request: Request, t: TranslatorDep, service: FileStorageService = Depends(get_file_storage_service), -) -> RedirectResponse | StreamingResponse: +) -> Response: try: - download = await service.download(file_id) - except StoredFileNotFoundError as exc: - raise not_found(t) from exc + download = await service.download(file_id, open_body=False) + if isinstance(download, RedirectDownload): + return RedirectResponse(url=download.url, status_code=status.HTTP_302_FOUND) - if isinstance(download, RedirectDownload): - return RedirectResponse(url=download.url, status_code=status.HTTP_302_FOUND) - - assert isinstance(download, StreamDownload) - row = download.file - return StreamingResponse( - download.body, - media_type=row.content_type, - headers={ - "Content-Disposition": f'attachment; filename="{row.filename}"', - "Content-Length": str(row.size_bytes), - "ETag": f'"{row.checksum_sha256}"', - }, - ) + assert isinstance(download, StreamDownload) + row = download.file + etag = etag_for(row) + return await file_body_response( + service.backend, + row, + request, + headers={ + "Content-Disposition": f'attachment; filename="{row.filename}"', + "ETag": etag, + }, + etag=etag, + ) + except (StoredFileNotFoundError, StorageNotFoundError) as exc: + raise not_found(t) from exc @router.post( diff --git a/modules/file_storage/file_storage/locales/en.json b/modules/file_storage/file_storage/locales/en.json index d5f2e288..5f8b9e10 100644 --- a/modules/file_storage/file_storage/locales/en.json +++ b/modules/file_storage/file_storage/locales/en.json @@ -51,6 +51,9 @@ "deleted_one": "“{name}” deleted", "deleted_other": "{count} files deleted", "delete_failed": "Failed to delete file", + "already_deleted_one": "This file was already deleted", + "already_deleted_other": "These {count} files were already deleted", + "deleted_partial": "{deleted} of {total} files deleted; the rest were already gone", "made_public": "“{name}” is now public", "made_private": "“{name}” is now private", "visibility_failed": "Could not change visibility", diff --git a/modules/file_storage/file_storage/pages/Browse.tsx b/modules/file_storage/file_storage/pages/Browse.tsx index f1500728..e1070817 100644 --- a/modules/file_storage/file_storage/pages/Browse.tsx +++ b/modules/file_storage/file_storage/pages/Browse.tsx @@ -125,9 +125,17 @@ function Browse() { // of a selection can already be gone, and naming the first file the user // ticked would credit a deletion that did not happen. const removed = ids.length === 1 ? files.find((f) => f.id === ids[0]) : undefined; - toast.success( - t(keys.file_storage.toasts.deleted, { count: deleted, name: removed?.filename ?? '' }), - ); + const requested = selectedIds.length; + if (deleted === 0) { + // A stale tab: every selected file was removed elsewhere already. + toast.info(t(keys.file_storage.toasts.already_deleted, { count: requested })); + } else if (deleted < requested) { + toast.success(t(keys.file_storage.toasts.deleted_partial, { deleted, total: requested })); + } else { + toast.success( + t(keys.file_storage.toasts.deleted, { count: deleted, name: removed?.filename ?? '' }), + ); + } select([]); router.reload({ only: RELOAD_PROPS }); } catch { diff --git a/modules/file_storage/file_storage/ranges.py b/modules/file_storage/file_storage/ranges.py new file mode 100644 index 00000000..4e4b0507 --- /dev/null +++ b/modules/file_storage/file_storage/ranges.py @@ -0,0 +1,123 @@ +"""HTTP ``Range`` support for streamed file bodies (RFC 9110 § 14). + +Only a single byte range is honoured — ``bytes=a-b``, ``bytes=a-`` and the +suffix form ``bytes=-n`` — which is all COG/PMTiles readers send. Anything +else (multiple ranges, another unit, garbage) is ignored and the whole object +is sent with ``200``, as RFC 9110 permits. A well-formed range that starts at +or past the end of the object is ``416`` with ``Content-Range: bytes */size``. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Final, Literal + +from fastapi import Request, Response, status +from fastapi.responses import StreamingResponse + +from file_storage.contracts.service import StorageBackend +from file_storage.models import StoredFile + +# 19 digits bounds every value below 2**63; longer numbers are treated as +# malformed (full 200) rather than parsed into arbitrarily large ints. +_SPEC = re.compile(r"^\s*(\d{0,19})\s*-\s*(\d{0,19})\s*$") + +UNSATISFIABLE: Final = "unsatisfiable" + + +@dataclass(frozen=True) +class ByteRange: + """``start`` through ``end``, both inclusive, already clamped to the object.""" + + start: int + end: int + + @property + def length(self) -> int: + return self.end - self.start + 1 + + +def parse_range(header: str | None, size: int) -> ByteRange | Literal["unsatisfiable"] | None: + """Resolve a ``Range`` header against an object of ``size`` bytes. + + ``None`` means "ignore it and send everything" (absent, malformed, multi-range + or a unit other than bytes); :data:`UNSATISFIABLE` means ``416``. + """ + if not header: + return None + unit, sep, spec = header.partition("=") + if not sep or unit.strip().lower() != "bytes" or "," in spec: + return None + match = _SPEC.match(spec) + if match is None: + return None + first, last = match.groups() + if not first: + if not last: + return None + suffix = int(last) + if suffix == 0 or size == 0: + return UNSATISFIABLE + return ByteRange(start=max(0, size - suffix), end=size - 1) + start = int(first) + if last and int(last) < start: + return None # syntactically invalid: RFC 9110 says ignore the header + if start >= size: + return UNSATISFIABLE + end = min(int(last), size - 1) if last else size - 1 + return ByteRange(start=start, end=end) + + +def _range_header(request: Request, etag: str) -> str | None: + """The ``Range`` to honour, dropped when an ``If-Range`` validator fails.""" + if_range = request.headers.get("if-range") + if if_range is not None and if_range.strip() != etag: + # A stale (or date-based) validator: the client's partial copy may be + # of another version, so it gets the whole current object instead. + return None + return request.headers.get("range") + + +async def file_body_response( + backend: StorageBackend, + row: StoredFile, + request: Request, + *, + headers: Mapping[str, str], + etag: str, +) -> Response: + """Stream ``row`` as ``200``, or the requested slice as ``206`` / ``416``. + + ``headers`` are the caller's headers for the full response, minus + ``Content-Length``, which is set here. The backend is opened only for the + response actually served, so a ranged request never opens the full object. + Raises :class:`StorageNotFoundError` if the object is gone. + """ + size = row.size_bytes + get_range = getattr(backend, "get_range", None) + parsed = parse_range(_range_header(request, etag), size) if get_range else None + out = dict(headers) + if get_range is not None: + out["Accept-Ranges"] = "bytes" + + if parsed is None: + body = await backend.get(row.key) + out["Content-Length"] = str(size) + return StreamingResponse(body, media_type=row.content_type, headers=out) + + if parsed == UNSATISFIABLE: + out["Content-Range"] = f"bytes */{size}" + return Response(status_code=status.HTTP_416_RANGE_NOT_SATISFIABLE, headers=out) + + assert isinstance(parsed, ByteRange) and get_range is not None + ranged = await get_range(row.key, parsed.start, parsed.end) + out["Content-Range"] = f"bytes {parsed.start}-{parsed.end}/{size}" + out["Content-Length"] = str(parsed.length) + return StreamingResponse( + ranged, + status_code=status.HTTP_206_PARTIAL_CONTENT, + media_type=row.content_type, + headers=out, + ) diff --git a/modules/file_storage/file_storage/service.py b/modules/file_storage/file_storage/service.py index 31da1875..0b5ecbb4 100644 --- a/modules/file_storage/file_storage/service.py +++ b/modules/file_storage/file_storage/service.py @@ -55,7 +55,7 @@ class StreamDownload: """Response shape for backends that proxy bytes through the app.""" file: StoredFile - body: AsyncIterator[bytes] + body: AsyncIterator[bytes] | None # None when the caller opted out via open_body=False @dataclass @@ -184,7 +184,9 @@ async def get(self, file_id: uuid.UUID, *, platform: bool = False) -> StoredFile raise StoredFileNotFoundError(str(file_id)) return row - async def download(self, file_id: uuid.UUID, *, platform: bool = False) -> Download: + async def download( + self, file_id: uuid.UUID, *, platform: bool = False, open_body: bool = True + ) -> Download: """Return either a streamed body or a redirect URL. Dispatch on ``backend.supports_presigned_url`` so the service stays @@ -197,7 +199,7 @@ async def download(self, file_id: uuid.UUID, *, platform: bool = False) -> Downl row.key, self.settings.s3_presign_ttl_seconds ) return RedirectDownload(file=row, url=url) - body = await self.backend.get(row.key) + body = await self.backend.get(row.key) if open_body else None return StreamDownload(file=row, body=body) # ── Delete ─────────────────────────────────────────────────────── diff --git a/modules/file_storage/file_storage/serving.py b/modules/file_storage/file_storage/serving.py index 6f4bae9a..3316cb0e 100644 --- a/modules/file_storage/file_storage/serving.py +++ b/modules/file_storage/file_storage/serving.py @@ -6,15 +6,16 @@ from urllib.parse import quote from fastapi import HTTPException, Request, Response, status -from fastapi.responses import RedirectResponse, StreamingResponse +from fastapi.responses import RedirectResponse from file_storage import constants, thumbnails from file_storage.contracts.service import StorageNotFoundError from file_storage.models import StoredFile +from file_storage.ranges import file_body_response from file_storage.service import FileStorageService -def _etag(row: StoredFile, suffix: str = "") -> str: +def etag_for(row: StoredFile, suffix: str = "") -> str: return f'"{row.checksum_sha256}{suffix}"' @@ -66,7 +67,7 @@ async def thumbnail_response( ) -> Response: """Serve ``row``'s resized variant; 404 for non-images, 422 if undecodable.""" snapped = thumbnails.snap_width(width) - etag = _etag(row, f"-w{snapped}") + etag = etag_for(row, f"-w{snapped}") headers = _headers(etag, cache_control) if ( request is not None @@ -100,7 +101,7 @@ async def public_file_response( everything else may redirect to the backend's own URL. """ max_age = constants.PUBLIC_MAX_AGE_SECONDS - etag = _etag(row) + etag = etag_for(row) active = is_active_content(row.content_type) headers = _headers(etag, f"public, max-age={max_age}") if _not_modified(request, etag): @@ -119,15 +120,15 @@ async def public_file_response( ) try: - body = await service.backend.get(row.key) + return await file_body_response( + service.backend, + row, + request, + headers={ + **headers, + "Content-Disposition": content_disposition(row.filename, attachment=active), + }, + etag=etag, + ) except StorageNotFoundError as exc: raise not_found(t) from exc - return StreamingResponse( - body, - media_type=row.content_type, - headers={ - **headers, - "Content-Disposition": content_disposition(row.filename, attachment=active), - "Content-Length": str(row.size_bytes), - }, - ) diff --git a/modules/file_storage/tests/test_backend_registry.py b/modules/file_storage/tests/test_backend_registry.py index d5369aab..2a44fdcb 100644 --- a/modules/file_storage/tests/test_backend_registry.py +++ b/modules/file_storage/tests/test_backend_registry.py @@ -35,6 +35,11 @@ async def empty() -> AsyncIterator[bytes]: return empty() + async def get_range( # pragma: no cover - unused + self, key: str, start: int, end: int + ) -> AsyncIterator[bytes]: + return await self.get(key) + async def delete(self, key: str) -> None: ... async def exists(self, key: str) -> bool: return False diff --git a/modules/file_storage/tests/test_file_storage_range_download.py b/modules/file_storage/tests/test_file_storage_range_download.py new file mode 100644 index 00000000..9f2da528 --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_range_download.py @@ -0,0 +1,148 @@ +"""``Range`` → ``206`` on the download endpoints, filesystem backend (#453).""" + +from __future__ import annotations + +import httpx +from file_storage import constants + +API = constants.ROUTE_PREFIX_API +PAYLOAD = bytes(range(256)) * 800 # 204,800 bytes: several 64 KiB chunks +SIZE = len(PAYLOAD) +DISPOSITION = 'attachment; filename="tiles.pmtiles"' + + +async def _upload(client: httpx.AsyncClient, **form) -> dict: + resp = await client.post( + f"{API}{constants.PATH_UPLOAD}", + files={"file": ("tiles.pmtiles", PAYLOAD, "application/octet-stream")}, + data=form, + ) + assert resp.status_code == 201, resp.text + return resp.json() + + +async def _download(client: httpx.AsyncClient, **headers: str) -> httpx.Response: + body = await _upload(client) + return await client.get(f"{API}/files/{body['id']}/download", headers=headers) + + +async def test_no_range_is_a_full_200(authenticated_client): + # identity: GZip would otherwise compress the full body and drop the length. + resp = await _download(authenticated_client, **{"Accept-Encoding": "identity"}) + assert resp.status_code == 200 + assert resp.content == PAYLOAD + assert resp.headers["content-length"] == str(SIZE) + assert resp.headers["accept-ranges"] == "bytes" + assert "content-range" not in resp.headers + assert resp.headers["content-disposition"] == DISPOSITION + assert resp.headers["etag"].startswith('"') + + +async def test_first_16k_is_a_206_with_exact_bytes(authenticated_client): + resp = await _download(authenticated_client, Range="bytes=0-16383") + assert resp.status_code == 206 + assert resp.content == PAYLOAD[:16384] + assert resp.headers["content-length"] == "16384" + assert resp.headers["content-range"] == f"bytes 0-16383/{SIZE}" + assert resp.headers["accept-ranges"] == "bytes" + assert resp.headers["content-disposition"] == DISPOSITION + + +async def test_range_spanning_chunks(authenticated_client): + start, end = 65_000, 140_000 + resp = await _download(authenticated_client, Range=f"bytes={start}-{end}") + assert resp.status_code == 206 + assert resp.content == PAYLOAD[start : end + 1] + + +async def test_suffix_range(authenticated_client): + resp = await _download(authenticated_client, Range="bytes=-100") + assert resp.status_code == 206 + assert resp.content == PAYLOAD[-100:] + assert resp.headers["content-range"] == f"bytes {SIZE - 100}-{SIZE - 1}/{SIZE}" + + +async def test_open_ended_range(authenticated_client): + resp = await _download(authenticated_client, Range="bytes=200000-") + assert resp.status_code == 206 + assert resp.content == PAYLOAD[200000:] + assert resp.headers["content-length"] == str(SIZE - 200000) + + +async def test_end_past_eof_is_clamped(authenticated_client): + resp = await _download(authenticated_client, Range="bytes=204700-999999999") + assert resp.status_code == 206 + assert resp.content == PAYLOAD[204700:] + assert resp.headers["content-range"] == f"bytes 204700-204799/{SIZE}" + + +async def test_unsatisfiable_range_is_416(authenticated_client): + resp = await _download(authenticated_client, Range=f"bytes={SIZE}-") + assert resp.status_code == 416 + assert resp.headers["content-range"] == f"bytes */{SIZE}" + assert resp.content == b"" + + +async def test_multi_range_and_garbage_fall_back_to_full_200(authenticated_client): + for header in ("bytes=0-1,5-6", "bytes=9-2", "lines=1-2", "bytes=abc"): + resp = await _download(authenticated_client, Range=header) + assert resp.status_code == 200, header + assert resp.content == PAYLOAD + + +async def test_stale_if_range_sends_everything(authenticated_client): + body = await _upload(authenticated_client) + url = f"{API}/files/{body['id']}/download" + etag = (await authenticated_client.get(url)).headers["etag"] + + hit = await authenticated_client.get(url, headers={"Range": "bytes=0-9", "If-Range": etag}) + assert hit.status_code == 206 + assert hit.content == PAYLOAD[:10] + stale = await authenticated_client.get( + url, headers={"Range": "bytes=0-9", "If-Range": '"other"'} + ) + assert stale.status_code == 200 + assert stale.content == PAYLOAD + + +async def test_range_still_requires_auth(authenticated_client, client): + body = await _upload(authenticated_client) + resp = await client.get(f"{API}/files/{body['id']}/download", headers={"Range": "bytes=0-9"}) + assert resp.status_code in (401, 302, 403) + + +async def test_public_route_serves_ranges_with_its_headers(authenticated_client, client): + body = await _upload(authenticated_client, public="true") + url = f"{API}/public/{body['id']}" + resp = await client.get(url, headers={"Range": "bytes=-16384"}) + assert resp.status_code == 206 + assert resp.content == PAYLOAD[-16384:] + assert resp.headers["cache-control"].startswith("public, max-age=") + assert resp.headers["x-content-type-options"] == "nosniff" + assert "sandbox" in resp.headers["content-security-policy"] + assert resp.headers["content-disposition"].startswith("inline") + + bad = await client.get(url, headers={"Range": "bytes=999999-"}) + assert bad.status_code == 416 + assert bad.headers["content-range"] == f"bytes */{SIZE}" + + +async def test_ranged_download_never_opens_the_full_body(authenticated_client, monkeypatch): + from file_storage.backends.filesystem import FilesystemBackend + + body = await _upload(authenticated_client) + opened: list[str] = [] + real_get = FilesystemBackend.get + + async def spy_get(self, key): + opened.append(key) + return await real_get(self, key) + + monkeypatch.setattr(FilesystemBackend, "get", spy_get) + url = f"{API}/files/{body['id']}/download" + resp = await authenticated_client.get(url, headers={"Range": "bytes=0-9"}) + assert resp.status_code == 206 + assert opened == [] + full = await authenticated_client.get(url, headers={"Accept-Encoding": "identity"}) + assert full.status_code == 200 + assert len(opened) == 1 diff --git a/modules/file_storage/tests/test_file_storage_range_parse.py b/modules/file_storage/tests/test_file_storage_range_parse.py new file mode 100644 index 00000000..bbd7f729 --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_range_parse.py @@ -0,0 +1,63 @@ +"""``Range`` header parsing (#453): single ranges only, RFC 9110 semantics.""" + +from __future__ import annotations + +import pytest +from file_storage.ranges import UNSATISFIABLE, ByteRange, parse_range + +SIZE = 1000 + + +@pytest.mark.parametrize( + ("header", "expected"), + [ + ("bytes=0-16383", ByteRange(0, 999)), # end clamped to size - 1 + ("bytes=0-0", ByteRange(0, 0)), + ("bytes=10-19", ByteRange(10, 19)), + ("bytes=990-", ByteRange(990, 999)), + ("bytes=-10", ByteRange(990, 999)), + ("bytes=-5000", ByteRange(0, 999)), # suffix longer than the object + ("BYTES = 5 - 9 ", ByteRange(5, 9)), # unit is case-insensitive + ("bytes=999-999", ByteRange(999, 999)), + ], +) +def test_satisfiable_ranges(header, expected): + assert parse_range(header, SIZE) == expected + + +@pytest.mark.parametrize( + "header", ["bytes=1000-", "bytes=1000-2000", "bytes=-0", "bytes=5000-6000"] +) +def test_unsatisfiable_ranges(header): + assert parse_range(header, SIZE) == UNSATISFIABLE + + +def test_any_range_of_an_empty_object_is_unsatisfiable(): + assert parse_range("bytes=0-", 0) == UNSATISFIABLE + assert parse_range("bytes=-1", 0) == UNSATISFIABLE + + +@pytest.mark.parametrize( + "header", + [ + None, + "", + "bytes=0-1,5-6", # multi-range: ignored, full 200 + "items=0-1", + "bytes", + "bytes=", + "bytes=-", + "bytes=5-1", # last < first is invalid syntax + "bytes=-1-2", + "bytes=a-b", + "bytes=+1-2", + "bytes=0x10-", + "bytes=" + "9" * 20 + "-", # absurdly long number + ], +) +def test_ignored_headers(header): + assert parse_range(header, SIZE) is None + + +def test_byte_range_length_is_inclusive(): + assert ByteRange(0, 16383).length == 16384 diff --git a/modules/file_storage/tests/test_file_storage_timestamps.py b/modules/file_storage/tests/test_file_storage_timestamps.py new file mode 100644 index 00000000..047cddff --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_timestamps.py @@ -0,0 +1,36 @@ +"""GH #468: API timestamps carry a UTC offset even on SQLite. + +Without one the browser parses ``2026-10-14T10:40:52`` as local time, so a +file uploaded a moment ago shows "2h ago" at UTC+2. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import httpx +from file_storage import constants + + +def _assert_utc(stamp: str) -> None: + assert stamp.endswith(("+00:00", "Z")), stamp + assert abs(datetime.now(UTC) - datetime.fromisoformat(stamp)) < timedelta(minutes=5) + + +async def test_list_and_detail_timestamps_have_utc_offset( + authenticated_client: httpx.AsyncClient, +): + upload = await authenticated_client.post( + f"{constants.ROUTE_PREFIX_API}{constants.PATH_UPLOAD}", + files={"file": ("when.txt", b"x", "text/plain")}, + ) + assert upload.status_code == 201, upload.text + file_id = upload.json()["id"] + + listing = await authenticated_client.get(f"{constants.ROUTE_PREFIX_API}{constants.PATH_FILES}") + assert listing.status_code == 200 + _assert_utc(listing.json()["items"][0]["created_at"]) + + detail = await authenticated_client.get(f"{constants.ROUTE_PREFIX_API}/files/{file_id}") + assert detail.status_code == 200 + _assert_utc(detail.json()["created_at"]) diff --git a/modules/file_storage/tests/test_filesystem_backend.py b/modules/file_storage/tests/test_filesystem_backend.py index 40a75bb0..c431c14a 100644 --- a/modules/file_storage/tests/test_filesystem_backend.py +++ b/modules/file_storage/tests/test_filesystem_backend.py @@ -65,3 +65,28 @@ async def test_two_char_shard_layout(tmp_path): await backend.put("ab/file.bin", _bytes_stream(b"x"), content_type="text/plain", size=1) # Stored at /ab/ab/file.bin (2-char shard prefix from key[:2]) assert (tmp_path / "ab" / "ab" / "file.bin").exists() + + +async def _drain(stream: AsyncIterator[bytes]) -> list[bytes]: + return [chunk async for chunk in stream] + + +async def test_get_range_is_inclusive_and_chunked(tmp_path): + from file_storage import constants + + backend = FilesystemBackend(root=tmp_path) + data = bytes(range(256)) * 1000 + await backend.put("ab/big", _bytes_stream(data), content_type="x/y", size=len(data)) + + assert b"".join(await _drain(await backend.get_range("ab/big", 0, 16383))) == data[:16384] + assert b"".join(await _drain(await backend.get_range("ab/big", 7, 7))) == data[7:8] + chunks = await _drain(await backend.get_range("ab/big", 1000, 200_000)) + assert b"".join(chunks) == data[1000:200_001] + assert len(chunks) > 1 + assert max(map(len, chunks)) <= constants.DEFAULT_CHUNK_SIZE + + +async def test_get_range_missing_raises_not_found(tmp_path): + backend = FilesystemBackend(root=tmp_path) + with pytest.raises(StorageNotFoundError): + await backend.get_range("missing/key", 0, 1) diff --git a/modules/file_storage/tests/test_s3_backend.py b/modules/file_storage/tests/test_s3_backend.py index 50cc3397..c7c7c303 100644 --- a/modules/file_storage/tests/test_s3_backend.py +++ b/modules/file_storage/tests/test_s3_backend.py @@ -82,7 +82,7 @@ async def test_s3_put_get_delete_roundtrip(s3_settings): size=len(payload), ) assert await backend.exists("2026/04/19/abc.bin") is True - out = await _drain(backend.get("2026/04/19/abc.bin")) + out = await _drain(await backend.get("2026/04/19/abc.bin")) assert out == payload await backend.delete("2026/04/19/abc.bin") @@ -91,8 +91,78 @@ async def test_s3_put_get_delete_roundtrip(s3_settings): async def test_s3_get_missing_raises_not_found(s3_settings): backend = build_backend(s3_settings) + # Raised by the awaited call itself, before any response could start. with pytest.raises(StorageNotFoundError): - await _drain(backend.get("missing")) + await backend.get("missing") + with pytest.raises(StorageNotFoundError): + await backend.get_range("missing", 0, 9) + + +_BIG = bytes(range(256)) * 1000 # 256,000 bytes: four 64 KiB reads + + +async def _put(backend, key: str, data: bytes) -> None: + await backend.put( + key, _bytes_stream(data), content_type="application/octet-stream", size=len(data) + ) + + +@pytest.fixture +def body_reads(monkeypatch) -> list[int | None]: + """Record the ``amt`` of every ``StreamingBody.read`` (``None`` = read-all).""" + from aiobotocore.response import StreamingBody + + calls: list[int | None] = [] + real = StreamingBody.read + + async def spy(self, amt=None): + calls.append(amt) + return await real(self, amt) + + monkeypatch.setattr(StreamingBody, "read", spy) + return calls + + +async def test_s3_get_streams_in_bounded_reads(s3_settings, body_reads): + # #453: the body used to be read whole before the first byte was sent. + backend = build_backend(s3_settings) + await _put(backend, "big.bin", _BIG) + chunks = [chunk async for chunk in await backend.get("big.bin")] + assert b"".join(chunks) == _BIG + assert len(chunks) > 1 + assert max(map(len, chunks)) <= constants.DEFAULT_CHUNK_SIZE + assert body_reads and None not in body_reads + assert set(body_reads) == {constants.DEFAULT_CHUNK_SIZE} + + +async def test_s3_get_range_is_inclusive_and_streamed(s3_settings, body_reads): + backend = build_backend(s3_settings) + await _put(backend, "big.bin", _BIG) + assert await _drain(await backend.get_range("big.bin", 0, 16383)) == _BIG[:16384] + assert await _drain(await backend.get_range("big.bin", 5, 5)) == _BIG[5:6] + tail = await _drain(await backend.get_range("big.bin", 100_000, len(_BIG) - 1)) + assert tail == _BIG[100_000:] + assert None not in body_reads + + +async def test_s3_closing_stream_early_releases_body(s3_settings, monkeypatch): + from aiobotocore.response import StreamingBody + + closed: list[bool] = [] + monkeypatch.setattr(StreamingBody, "close", lambda self: closed.append(True), raising=False) + backend = build_backend(s3_settings) + await _put(backend, "big.bin", _BIG) + stream = await backend.get("big.bin") + first = await anext(aiter(stream)) + assert len(first) <= constants.DEFAULT_CHUNK_SIZE + await stream.aclose() # what Starlette does when the client disconnects + assert closed == [True] + + +async def test_s3_get_empty_object(s3_settings): + backend = build_backend(s3_settings) + await _put(backend, "empty.bin", b"") + assert await _drain(await backend.get("empty.bin")) == b"" async def test_s3_presigned_url_format(s3_settings): diff --git a/modules/permissions/permissions/models.py b/modules/permissions/permissions/models.py index 105303c8..965955b0 100644 --- a/modules/permissions/permissions/models.py +++ b/modules/permissions/permissions/models.py @@ -25,7 +25,8 @@ from fastapi_users_db_sqlalchemy.generics import GUID, now_utc from simple_module_db.base import create_module_base from simple_module_db.mixins import AuditMixin -from sqlalchemy import DateTime, Index +from simple_module_db.types import AwareDateTime +from sqlalchemy import Index from sqlmodel import Field # Provider is auto-detected from SM_DATABASE_URL (falls back to SQLite). @@ -43,7 +44,7 @@ class RolePermission(Base, AuditMixin, table=True): # ty: ignore[unsupported-ba permission_key: str = Field(max_length=128, primary_key=True) assigned_at: datetime = Field( default_factory=now_utc, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), ) assigned_by: str | None = Field(default=None, max_length=255) @@ -62,7 +63,7 @@ class UserPermission(Base, AuditMixin, table=True): # ty: ignore[unsupported-ba permission_key: str = Field(max_length=128, primary_key=True) assigned_at: datetime = Field( default_factory=now_utc, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), ) assigned_by: str | None = Field(default=None, max_length=255) diff --git a/modules/tenants/tenants/models.py b/modules/tenants/tenants/models.py index 7375a51e..c135e3ee 100644 --- a/modules/tenants/tenants/models.py +++ b/modules/tenants/tenants/models.py @@ -14,7 +14,8 @@ from simple_module_db import is_valid_tenant_id from simple_module_db.base import create_module_base from simple_module_db.mixins import AuditMixin -from sqlalchemy import Column, DateTime, Index, UniqueConstraint, text +from simple_module_db.types import AwareDateTime +from sqlalchemy import Column, Index, UniqueConstraint, text from sqlalchemy.orm import validates from sqlmodel import Field @@ -103,7 +104,7 @@ class Invitation(Base, AuditMixin, table=True): # ty: ignore[unsupported-base] email: str = Field(max_length=MAX_EMAIL_LEN) role: str = Field(default=MembershipRole.MEMBER, max_length=20) token_hash: str = Field(max_length=64, unique=True, index=True) - expires_at: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False)) + expires_at: datetime = Field(sa_column=Column(AwareDateTime(), nullable=False)) accepted_at: datetime | None = Field( - default=None, sa_column=Column(DateTime(timezone=True), nullable=True) + default=None, sa_column=Column(AwareDateTime(), nullable=True) ) diff --git a/modules/users/tests/test_token_api.py b/modules/users/tests/test_token_api.py index 0c012eca..c32afb59 100644 --- a/modules/users/tests/test_token_api.py +++ b/modules/users/tests/test_token_api.py @@ -71,6 +71,27 @@ async def test_inactive_user_returns_401(self, anon_client, users_db): ) assert resp.status_code == 401 + @pytest.mark.anyio + async def test_legacy_hash_is_upgraded_on_token_login(self, anon_client, users_db): + from pwdlib.hashers.bcrypt import BcryptHasher + + legacy = BcryptHasher().hash("SecurePass1!") + assert legacy.startswith("$2") + user = await _seed_user(users_db, email="legacy@example.com") + user.hashed_password = legacy + await users_db.commit() + + for _ in range(2): # second login proves the upgraded hash still verifies + resp = await anon_client.post( + "/api/users/auth/token", + json={"email": "legacy@example.com", "password": "SecurePass1!"}, + ) + assert resp.status_code == 200 + + await users_db.refresh(user) + assert user.hashed_password != legacy + assert user.hashed_password.startswith("$argon2") + @pytest.mark.anyio async def test_valid_credentials_returns_token_pair(self, anon_client, users_db): await _seed_user(users_db) diff --git a/modules/users/users/auth_local/token_api.py b/modules/users/users/auth_local/token_api.py index 6d1af5b5..9666f264 100644 --- a/modules/users/users/auth_local/token_api.py +++ b/modules/users/users/auth_local/token_api.py @@ -79,9 +79,15 @@ async def token_login( helper.verify_and_update(body.password, _DUMMY_HASH) raise HTTPException(status_code=401, detail="Invalid credentials") - verified, _ = helper.verify_and_update(body.password, user.hashed_password) + verified, updated_hash = helper.verify_and_update(body.password, user.hashed_password) if not verified: raise HTTPException(status_code=401, detail="Invalid credentials") + if updated_hash is not None: + # A legacy-scheme hash (e.g. bcrypt) was accepted and re-hashed with the + # current scheme; persist it, as ``UserManager.authenticate`` does for + # the session login. ``get_db`` commits since this is a pending write. + user.hashed_password = updated_hash + await db.flush() settings = request.app.state.users.settings return await _create_token_pair(db, user.id, settings, user.session_version) diff --git a/modules/users/users/models/user.py b/modules/users/users/models/user.py index 347bcbae..75ae3af4 100644 --- a/modules/users/users/models/user.py +++ b/modules/users/users/models/user.py @@ -16,7 +16,8 @@ from fastapi_users_db_sqlalchemy.generics import GUID from simple_module_db.mixins import AuditMixin -from sqlalchemy import DateTime, Index, false, text +from simple_module_db.types import AwareDateTime +from sqlalchemy import Index, false, text from sqlmodel import Field, Relationship from users.models._base import Base @@ -58,11 +59,11 @@ class User(Base, AuditMixin, table=True): # ty: ignore[unsupported-base] full_name: str | None = Field(default=None, max_length=255) disabled_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), ) last_login_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), index=True, ) # Set when an admin created this account on someone's behalf; ``None`` for @@ -74,7 +75,7 @@ class User(Base, AuditMixin, table=True): # ty: ignore[unsupported-base] # lifetime. invited_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), ) # Bumped by "sign out everywhere". Browser auth is a signed cookie, not a # server-side session store, so there is no row to delete for a device that diff --git a/modules/users/users/models/user_role.py b/modules/users/users/models/user_role.py index ff81b555..d2d7fee7 100644 --- a/modules/users/users/models/user_role.py +++ b/modules/users/users/models/user_role.py @@ -4,7 +4,8 @@ from datetime import datetime from fastapi_users_db_sqlalchemy.generics import GUID, now_utc -from sqlalchemy import DateTime, Index +from simple_module_db.types import AwareDateTime +from sqlalchemy import Index from sqlmodel import Field from users.models._base import Base @@ -29,7 +30,7 @@ class UserRole(Base, table=True): # ty: ignore[unsupported-base] ) assigned_at: datetime = Field( default_factory=now_utc, - sa_type=DateTime(timezone=True), + sa_type=AwareDateTime(), ) assigned_by: str | None = Field(default=None, max_length=255) diff --git a/packages/i18n/src/generated-resources.ts b/packages/i18n/src/generated-resources.ts index 60a46de2..8a3fc4c6 100644 --- a/packages/i18n/src/generated-resources.ts +++ b/packages/i18n/src/generated-resources.ts @@ -406,9 +406,12 @@ export default { 'file_storage.table.type': '', 'file_storage.table.uploaded_by': '', 'file_storage.table.when': '', + 'file_storage.toasts.already_deleted_one': '', + 'file_storage.toasts.already_deleted_other': '', 'file_storage.toasts.delete_failed': '', 'file_storage.toasts.deleted_one': '', 'file_storage.toasts.deleted_other': '', + 'file_storage.toasts.deleted_partial': '', 'file_storage.toasts.made_private': '', 'file_storage.toasts.made_public': '', 'file_storage.toasts.upload_failed': '', diff --git a/packages/i18n/src/keys.generated.ts b/packages/i18n/src/keys.generated.ts index c06942c0..275aa76a 100644 --- a/packages/i18n/src/keys.generated.ts +++ b/packages/i18n/src/keys.generated.ts @@ -522,10 +522,14 @@ export const keys = { when: 'file_storage.table.when', }, toasts: { + already_deleted: 'file_storage.toasts.already_deleted', + already_deleted_one: 'file_storage.toasts.already_deleted_one', + already_deleted_other: 'file_storage.toasts.already_deleted_other', delete_failed: 'file_storage.toasts.delete_failed', deleted: 'file_storage.toasts.deleted', deleted_one: 'file_storage.toasts.deleted_one', deleted_other: 'file_storage.toasts.deleted_other', + deleted_partial: 'file_storage.toasts.deleted_partial', made_private: 'file_storage.toasts.made_private', made_public: 'file_storage.toasts.made_public', upload_failed: 'file_storage.toasts.upload_failed',