From 255ec2aba77a7c5e5079112dea681a49ace8e603 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Apr 2026 14:16:33 +0000 Subject: [PATCH 1/3] Drop unused service Protocols for single-impl modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The IProductService, IPermissionService, IBackgroundTaskService, and ISettingService Protocols were never imported outside their defining modules — deps.py wires the concrete service, endpoints type-hint against the concrete class, and there are no alternate implementations. Keeping them was ceremony: every signature change required editing two files, and readers had to chase the Protocol to find the real code. Keep file_storage.StorageBackend: it has a registry, two shipped backends (filesystem, S3), and tests that mock against it — a real extension point that earns the abstraction. - Delete the four contracts/service.py Protocol files and drop the re-exports from the contracts packages' __init__.py. - SettingsAccessor now type-hints against SettingService directly, under a TYPE_CHECKING guard so contracts/ stays import-cycle-free. - Update the scaffold (scripts/_templates_contracts.py, new_module.py) to stop generating contracts/service.py for new modules. - Document the rule in docs/module-authoring.md: ship a Protocol only for real extension points. --- CLAUDE.md | 2 +- docs/framework-conventions.md | 2 +- docs/module-authoring.md | 14 ++++++ .../background_tasks/contracts/service.py | 32 ------------ .../permissions/contracts/__init__.py | 2 - .../permissions/contracts/service.py | 33 ------------ .../products/products/contracts/__init__.py | 2 - .../products/products/contracts/service.py | 17 ------- .../settings/settings/contracts/__init__.py | 2 - .../settings/settings/contracts/accessor.py | 12 +++-- .../settings/settings/contracts/service.py | 50 ------------------- scripts/_templates_contracts.py | 39 +++------------ scripts/new_module.py | 3 +- 13 files changed, 32 insertions(+), 178 deletions(-) delete mode 100644 modules/background_tasks/background_tasks/contracts/service.py delete mode 100644 modules/permissions/permissions/contracts/service.py delete mode 100644 modules/products/products/contracts/service.py delete mode 100644 modules/settings/settings/contracts/service.py diff --git a/CLAUDE.md b/CLAUDE.md index c735eb03..a40b7189 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,7 +39,7 @@ This is a **modular-monolith framework**. There is no host–module API boundary modules/// ├── module.py # ModuleBase subclass with meta = ModuleMeta(...) ├── models.py # SQLModel tables -├── contracts/ # SQLModel DTOs + Protocol interfaces (public surface) +├── contracts/ # SQLModel DTOs (public surface) — add a Protocol only for real extension points ├── service.py # business logic ├── deps.py # FastAPI dependencies ├── endpoints/api.py # REST (JSON) diff --git a/docs/framework-conventions.md b/docs/framework-conventions.md index 2b7e5789..4edf549b 100644 --- a/docs/framework-conventions.md +++ b/docs/framework-conventions.md @@ -13,7 +13,7 @@ modules// ├── models.py # SQLModel tables (optional) ├── service.py # business logic ├── deps.py # FastAPI dependencies - ├── contracts/ # SQLModel DTOs + Protocol interfaces (public) + ├── contracts/ # SQLModel DTOs (public surface) ├── endpoints/ │ ├── api.py # REST endpoints (JSON) │ └── views.py # Inertia view endpoints diff --git a/docs/module-authoring.md b/docs/module-authoring.md index 5bdd5797..9b219dc4 100644 --- a/docs/module-authoring.md +++ b/docs/module-authoring.md @@ -18,10 +18,24 @@ my-module/ │ ├── pages/ # Inertia TSX pages (optional) │ ├── templates/ # Jinja2 templates (optional) │ ├── static/dist/ # pre-built frontend assets (optional) +│ ├── contracts/schemas.py # SQLModel DTOs — the public surface │ └── contracts/events.py # domain events (optional) └── tests/ ``` +### Service types: concrete class, not Protocol + +Export the concrete service class from `.service` and have consumers +type-hint against it. Do **not** ship a `contracts/service.py` with an +`IFooService` Protocol by default — it's dead boilerplate when there's only +one implementation. + +Add a Protocol only when the module is a real extension point with multiple +interchangeable implementations that an operator can swap at runtime. The +canonical example is `file_storage.StorageBackend`: it has a registry, two +shipped implementations (`FilesystemBackend`, `S3Backend`), and tests that +mock against the Protocol. If none of those apply to your module, skip it. + ### Minimal `pyproject.toml` ```toml diff --git a/modules/background_tasks/background_tasks/contracts/service.py b/modules/background_tasks/background_tasks/contracts/service.py deleted file mode 100644 index 6966af70..00000000 --- a/modules/background_tasks/background_tasks/contracts/service.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Public service interface for the BackgroundTasks module. - -Other modules depend on this Protocol, not on the concrete implementation. -""" - -from __future__ import annotations - -import uuid -from typing import Protocol - -from background_tasks.constants import TaskStatus -from background_tasks.contracts.schemas import ( - TaskExecutionDetail, - TaskExecutionListResponse, -) - - -class IBackgroundTaskService(Protocol): - """Surface exposed to callers outside the module.""" - - async def list( - self, - *, - status: TaskStatus | None = None, - task_name: str | None = None, - page: int = 1, - per_page: int = 20, - ) -> TaskExecutionListResponse: ... - - async def get(self, execution_id: uuid.UUID) -> TaskExecutionDetail | None: ... - - async def retry(self, execution_id: uuid.UUID) -> TaskExecutionDetail: ... diff --git a/modules/permissions/permissions/contracts/__init__.py b/modules/permissions/permissions/contracts/__init__.py index cbdc8ffd..4ce2ecd1 100644 --- a/modules/permissions/permissions/contracts/__init__.py +++ b/modules/permissions/permissions/contracts/__init__.py @@ -9,10 +9,8 @@ UserPermissionsOut, UserPermissionsUpdate, ) -from permissions.contracts.service import IPermissionService __all__ = [ - "IPermissionService", "PermissionGroupOut", "RoleOut", "RolePermissionsOut", diff --git a/modules/permissions/permissions/contracts/service.py b/modules/permissions/permissions/contracts/service.py deleted file mode 100644 index 5ec7ef17..00000000 --- a/modules/permissions/permissions/contracts/service.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Permission service protocol — the public contract other modules depend on.""" - -from __future__ import annotations - -import uuid -from typing import Protocol - -from permissions.contracts.schemas import ( - PermissionGroupOut, - RolePermissionsOut, - UserPermissionsOut, -) - - -class IPermissionService(Protocol): - """Interface for permission-assignment operations.""" - - def list_registered_groups(self) -> list[PermissionGroupOut]: ... - async def get_role_permissions(self, role_id: uuid.UUID) -> RolePermissionsOut | None: ... - async def set_role_permissions( - self, - role_id: uuid.UUID, - keys: list[str], - assigned_by: str | None = None, - ) -> RolePermissionsOut | None: ... - async def get_user_permissions(self, user_id: uuid.UUID) -> UserPermissionsOut | None: ... - async def set_user_permissions( - self, - user_id: uuid.UUID, - keys: list[str], - assigned_by: str | None = None, - ) -> UserPermissionsOut | None: ... - async def resolve_effective_permissions(self, user_id: uuid.UUID) -> set[str]: ... diff --git a/modules/products/products/contracts/__init__.py b/modules/products/products/contracts/__init__.py index 81a03e11..a33748e9 100644 --- a/modules/products/products/contracts/__init__.py +++ b/modules/products/products/contracts/__init__.py @@ -10,10 +10,8 @@ ProductOut, ProductUpdate, ) -from products.contracts.service import IProductService __all__ = [ - "IProductService", "ProductCreate", "ProductCreated", "ProductDeleted", diff --git a/modules/products/products/contracts/service.py b/modules/products/products/contracts/service.py deleted file mode 100644 index 5a1206db..00000000 --- a/modules/products/products/contracts/service.py +++ /dev/null @@ -1,17 +0,0 @@ -"""Product service protocol — the public contract other modules depend on.""" - -from __future__ import annotations - -from typing import Protocol - -from products.contracts.schemas import ProductCreate, ProductOut, ProductUpdate - - -class IProductService(Protocol): - """Interface for product operations.""" - - async def get_all(self) -> list[ProductOut]: ... - async def get_by_id(self, product_id: int) -> ProductOut | None: ... - async def create(self, data: ProductCreate) -> ProductOut: ... - async def update(self, product_id: int, data: ProductUpdate) -> ProductOut | None: ... - async def delete(self, product_id: int) -> bool: ... diff --git a/modules/settings/settings/contracts/__init__.py b/modules/settings/settings/contracts/__init__.py index 4e64a43e..f0957b23 100644 --- a/modules/settings/settings/contracts/__init__.py +++ b/modules/settings/settings/contracts/__init__.py @@ -10,10 +10,8 @@ SettingUpsert, SettingValueType, ) -from settings.contracts.service import ISettingService __all__ = [ - "ISettingService", "SettingCreate", "SettingDefinition", "SettingOut", diff --git a/modules/settings/settings/contracts/accessor.py b/modules/settings/settings/contracts/accessor.py index 2c933b2c..b814ff9c 100644 --- a/modules/settings/settings/contracts/accessor.py +++ b/modules/settings/settings/contracts/accessor.py @@ -21,7 +21,7 @@ async def handler(settings: SettingsDep): import json from collections.abc import Callable -from typing import Any, Final +from typing import TYPE_CHECKING, Any, Final from settings.constants import SYSTEM_SCOPE_ID from settings.contracts.registry import SettingsRegistry @@ -33,7 +33,9 @@ async def handler(settings: SettingsDep): SettingUpsert, SettingValueType, ) -from settings.contracts.service import ISettingService + +if TYPE_CHECKING: + from settings.service import SettingService class _Unset: @@ -82,7 +84,7 @@ def _cast_json(raw: str, default: Any) -> Any: class SettingsAccessor: - """Request-scoped facade over ``ISettingService``. + """Request-scoped facade over ``SettingService``. Bound to an optional ``user_id`` + ``tenant_id`` so ``get`` and its typed variants resolve via USER > TENANT > SYSTEM automatically. @@ -94,7 +96,7 @@ class SettingsAccessor: def __init__( self, - service: ISettingService, + service: SettingService, registry: SettingsRegistry | None = None, *, user_id: str | None = None, @@ -106,7 +108,7 @@ def __init__( self._tenant_id = tenant_id @property - def service(self) -> ISettingService: + def service(self) -> SettingService: return self._svc @property diff --git a/modules/settings/settings/contracts/service.py b/modules/settings/settings/contracts/service.py deleted file mode 100644 index f9dcbf9c..00000000 --- a/modules/settings/settings/contracts/service.py +++ /dev/null @@ -1,50 +0,0 @@ -"""Setting service protocol — the public contract other modules depend on.""" - -from __future__ import annotations - -from typing import Protocol - -from settings.contracts.schemas import ( - SettingCreate, - SettingOut, - SettingScope, - SettingUpdate, - SettingUpsert, -) - - -class ISettingService(Protocol): - """Interface for scoped key/value settings. - - Resolution precedence (high → low): USER > TENANT > SYSTEM. - """ - - async def list_all(self) -> list[SettingOut]: ... - async def list_by_scope(self, scope: SettingScope, scope_id: str = "") -> list[SettingOut]: ... - - async def get_by_id(self, setting_id: int) -> SettingOut | None: ... - async def get_scoped( - self, scope: SettingScope, scope_id: str, key: str - ) -> SettingOut | None: ... - async def resolve( - self, - key: str, - user_id: str | None = None, - tenant_id: str | None = None, - ) -> SettingOut | None: ... - async def get_resolved_value( - self, - key: str, - user_id: str | None = None, - tenant_id: str | None = None, - default: str | None = None, - ) -> str | None: ... - - async def create(self, data: SettingCreate) -> SettingOut: ... - async def update(self, setting_id: int, data: SettingUpdate) -> SettingOut | None: ... - async def upsert_scoped( - self, scope: SettingScope, scope_id: str, key: str, data: SettingUpsert - ) -> SettingOut: ... - - async def delete(self, setting_id: int) -> bool: ... - async def delete_scoped(self, scope: SettingScope, scope_id: str, key: str) -> bool: ... diff --git a/scripts/_templates_contracts.py b/scripts/_templates_contracts.py index 79f8c6f1..af71fbbb 100644 --- a/scripts/_templates_contracts.py +++ b/scripts/_templates_contracts.py @@ -1,4 +1,11 @@ -"""Template generators for the contracts/ sub-package (schemas + service protocol).""" +"""Template generators for the contracts/ sub-package (DTOs only). + +Modules do not ship a Protocol for their service by default. Consumers +type-hint against the concrete service class exported from +``.service`` — add a Protocol only when you expect multiple +implementations or an extension point (see ``file_storage.StorageBackend`` +for the pattern). +""" from __future__ import annotations @@ -14,13 +21,11 @@ def contracts_init(ctx: ScaffoldContext) -> str: {ctx.singular_class}Out, {ctx.singular_class}Update, ) - from {ctx.pkg}.contracts.service import I{ctx.singular_class}Service __all__ = [ "{ctx.singular_class}Create", "{ctx.singular_class}Out", "{ctx.singular_class}Update", - "I{ctx.singular_class}Service", ] ''' @@ -64,31 +69,3 @@ class {ctx.singular_class}Update(SQLModel): description: str | None = None is_active: bool | None = None ''' - - -def contracts_service(ctx: ScaffoldContext) -> str: - return f'''\ - """{ctx.singular_class} service protocol — the public contract other modules depend on.""" - - from __future__ import annotations - - from typing import Protocol - - from {ctx.pkg}.contracts.schemas import ( - {ctx.singular_class}Create, - {ctx.singular_class}Out, - {ctx.singular_class}Update, - ) - - - class I{ctx.singular_class}Service(Protocol): - """Interface for {ctx.singular} operations.""" - - async def get_all(self) -> list[{ctx.singular_class}Out]: ... - async def get_by_id(self, {ctx.singular}_id: int) -> {ctx.singular_class}Out | None: ... - async def create(self, data: {ctx.singular_class}Create) -> {ctx.singular_class}Out: ... - async def update( - self, {ctx.singular}_id: int, data: {ctx.singular_class}Update - ) -> {ctx.singular_class}Out | None: ... - async def delete(self, {ctx.singular}_id: int) -> bool: ... - ''' diff --git a/scripts/new_module.py b/scripts/new_module.py index 532d9b96..dffc684d 100644 --- a/scripts/new_module.py +++ b/scripts/new_module.py @@ -22,7 +22,7 @@ # Make sibling template modules importable when this file is run as a script. sys.path.insert(0, str(Path(__file__).resolve().parent)) -from _templates_contracts import contracts_init, contracts_service, schemas_py +from _templates_contracts import contracts_init, schemas_py from _templates_endpoints import api_py, views_py from _templates_py import ( ScaffoldContext, @@ -103,7 +103,6 @@ def scaffold_module(name: str) -> None: create_file(src_dir / "models.py", models_py(ctx)) create_file(src_dir / "contracts" / "__init__.py", contracts_init(ctx)) create_file(src_dir / "contracts" / "schemas.py", schemas_py(ctx)) - create_file(src_dir / "contracts" / "service.py", contracts_service(ctx)) create_file(src_dir / "service.py", service_py(ctx)) create_file(src_dir / "deps.py", deps_py(ctx)) create_file(src_dir / "endpoints" / "__init__.py", "") From deeb6ea16fe7f854fcc86acf8e7edcb8ffe60786 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Apr 2026 16:28:45 +0000 Subject: [PATCH 2/3] Publish TaskFailed and UserRegistered events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both events were declared in contracts/events.py but never emitted. Wire them up at the natural lifecycle hooks: - UserRegistered fires from UserManager.on_after_register using the EventBus on request.app.state.sm. Bootstrap/CLI flows pass request=None — those skip the publish instead of crashing. - TaskFailed fires from the Celery task_failure signal via a new bind_event_bus binder set in BackgroundTasksModule.on_startup. The signal runs on a sync Celery thread, so we hop back to the API loop with run_coroutine_threadsafe. In a standalone worker process the bind never runs and the publish is a no-op — that cross-process limitation is called out in the signals.py module docstring. Tests assert both events are emitted and that missing-bus paths remain no-ops. --- .../background_tasks/module.py | 9 +++ .../background_tasks/signals.py | 66 ++++++++++++++++- .../background_tasks/tests/test_signals.py | 70 +++++++++++++++++++ modules/users/tests/test_user_manager.py | 34 +++++++++ modules/users/users/manager.py | 20 ++++++ 5 files changed, 197 insertions(+), 2 deletions(-) diff --git a/modules/background_tasks/background_tasks/module.py b/modules/background_tasks/background_tasks/module.py index c98d5cb0..c54a4acd 100644 --- a/modules/background_tasks/background_tasks/module.py +++ b/modules/background_tasks/background_tasks/module.py @@ -77,12 +77,19 @@ def locale_dirs(self) -> dict[str, Path]: async def on_startup(self, app: FastAPI) -> None: """Build the Celery app, install signal handlers, hand-off to worker.""" + import asyncio + from background_tasks.celery_app import build_celery + from background_tasks.signals import bind_event_bus services = app.state.background_tasks # build_celery imports `signals` for side effects and runs # `autodiscover_tasks` across every installed module. services.celery = build_celery(services.settings) + # Let signal handlers hop onto the API loop to publish events. + # In a standalone worker process this bind never runs, so signals + # stay a silent no-op — that's the documented cross-process limit. + bind_event_bus(app.state.sm.event_bus, asyncio.get_running_loop()) logger.info( "BackgroundTasks: Celery app ready (broker=%s, queue=%s)", services.settings.broker_url, @@ -90,9 +97,11 @@ async def on_startup(self, app: FastAPI) -> None: ) async def on_shutdown(self, app: FastAPI) -> None: + from background_tasks.signals import unbind_event_bus from background_tasks.sync_db import dispose_sync_engine services = app.state.background_tasks if services.celery is not None: services.celery.close() + unbind_event_bus() dispose_sync_engine() diff --git a/modules/background_tasks/background_tasks/signals.py b/modules/background_tasks/background_tasks/signals.py index 70a687fd..fbea53ce 100644 --- a/modules/background_tasks/background_tasks/signals.py +++ b/modules/background_tasks/background_tasks/signals.py @@ -8,15 +8,24 @@ Signals are sync — see :mod:`.sync_db` for why we maintain a separate sync engine, and :mod:`._signal_support` for the shared helpers. + +``TaskFailed`` is also dispatched from :func:`on_task_failure` when an event +bus has been bound via :func:`bind_event_bus` — typically from the web +process's ``on_startup``. In a standalone Celery worker (separate process) +no bus is bound and the publish becomes a no-op; subscribers that need +cross-process notification should consume Celery events or poll the DB +directly. """ from __future__ import annotations +import asyncio import logging from collections.abc import Callable from typing import Any from celery import signals +from simple_module_core.events import EventBus from background_tasks._signal_support import ( coerce_args_kwargs, @@ -28,12 +37,47 @@ upsert_by_celery_id, ) from background_tasks.constants import DEFAULT_QUEUE, TaskStatus +from background_tasks.contracts.events import TaskFailed from background_tasks.models import TaskExecution from background_tasks.sync_db import sync_session logger = logging.getLogger(__name__) +_bus: EventBus | None = None +_loop: asyncio.AbstractEventLoop | None = None + + +def bind_event_bus(bus: EventBus, loop: asyncio.AbstractEventLoop) -> None: + """Bind an event bus + its running loop so signals can publish events. + + Signals fire on the Celery sync thread; `run_coroutine_threadsafe` + bridges back to ``loop`` so handlers run on the API event loop + regardless of which thread triggered the signal. + """ + global _bus, _loop + _bus = bus + _loop = loop + + +def unbind_event_bus() -> None: + """Drop the bound bus — called from ``on_shutdown`` so tests stay isolated.""" + global _bus, _loop + _bus = None + _loop = None + + +def _publish_from_signal(event: Any) -> None: + """Dispatch ``event`` onto the bound bus without blocking the signal thread.""" + if _bus is None or _loop is None: + return + try: + asyncio.run_coroutine_threadsafe(_bus.publish(event), _loop) + except RuntimeError: + # Loop has stopped (shutdown race). Swallow — the DB row is already written. + logger.debug("Event bus loop is not running; skipping %s", type(event).__name__) + + def _apply( handler: str, *, @@ -160,18 +204,36 @@ def on_task_failure( einfo: Any = None, **_k: Any, ) -> None: + task_name = task_name_of(sender) + exception_type = type(exception).__name__ if exception is not None else None + + published_id: list[Any] = [] + + def _capture_id(row: TaskExecution) -> None: + published_id.append(row.id) + _apply( "on_task_failure", celery_task_id=task_id, defaults={ - "task_name": task_name_of(sender), + "task_name": task_name, "status": TaskStatus.FAILED, "traceback": render_traceback(einfo, exception), - "exception_type": type(exception).__name__ if exception is not None else None, + "exception_type": exception_type, "finished_at": now_utc(), }, + after=_capture_id, ) + if published_id: + _publish_from_signal( + TaskFailed( + task_execution_id=published_id[0], + task_name=task_name, + exception_type=exception_type, + ) + ) + @signals.task_retry.connect def on_task_retry( diff --git a/modules/background_tasks/tests/test_signals.py b/modules/background_tasks/tests/test_signals.py index 8279b17d..26d0a5ca 100644 --- a/modules/background_tasks/tests/test_signals.py +++ b/modules/background_tasks/tests/test_signals.py @@ -7,23 +7,29 @@ from __future__ import annotations +import asyncio import uuid from collections.abc import Iterator from pathlib import Path import pytest +from background_tasks import signals as bg_signals from background_tasks import sync_db from background_tasks._signal_support import upsert_by_celery_id from background_tasks.constants import TaskStatus +from background_tasks.contracts.events import TaskFailed from background_tasks.models import TaskExecution from background_tasks.signals import ( + bind_event_bus, on_task_failure, on_task_prerun, on_task_publish, on_task_retry, on_task_revoked, on_task_success, + unbind_event_bus, ) +from simple_module_core.events import EventBus from sqlalchemy import select @@ -190,6 +196,70 @@ class _Task: assert row.finished_at is not None +class TestTaskFailedEvent: + @pytest.mark.asyncio + async def test_publishes_when_bus_bound(self, sync_sqlite: Path): + bus = EventBus() + received: list[TaskFailed] = [] + + async def _on_failed(event: TaskFailed) -> None: + received.append(event) + + bus.subscribe(TaskFailed, _on_failed) + loop = asyncio.get_running_loop() + bind_event_bus(bus, loop) + + try: + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.boom", + headers={"id": task_id, "task": "demo.boom"}, + body=([], {}, {}), + ) + + class _Task: + name = "demo.boom" + + class _Einfo: + traceback = "Traceback" + + on_task_failure( + sender=_Task, task_id=task_id, exception=ValueError("nope"), einfo=_Einfo() + ) + + # run_coroutine_threadsafe schedules the publish on this loop; + # yield long enough for the scheduled coroutine to complete. + await asyncio.sleep(0.05) + + assert len(received) == 1 + assert received[0].task_name == "demo.boom" + assert received[0].exception_type == "ValueError" + assert received[0].task_execution_id is not None + finally: + unbind_event_bus() + + def test_no_op_when_bus_unbound(self, sync_sqlite: Path): + """Without a bound bus (the standalone-worker case) the signal must + still record the failure row and not raise.""" + assert bg_signals._bus is None + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.boom", + headers={"id": task_id, "task": "demo.boom"}, + body=([], {}, {}), + ) + + class _Task: + name = "demo.boom" + + on_task_failure(sender=_Task, task_id=task_id, exception=RuntimeError("x"), einfo=None) + + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.status == TaskStatus.FAILED + + class TestUpsertHelper: def test_updates_existing_row_in_place(self, sync_sqlite: Path): factory = sync_db.get_sync_session_factory() diff --git a/modules/users/tests/test_user_manager.py b/modules/users/tests/test_user_manager.py index 7242ce11..284fdf74 100644 --- a/modules/users/tests/test_user_manager.py +++ b/modules/users/tests/test_user_manager.py @@ -112,6 +112,40 @@ async def test_accepts_valid_password(self, manager, fake_user): # --------------------------------------------------------------------------- +class TestOnAfterRegister: + @pytest.mark.anyio + async def test_publishes_user_registered(self, manager, fake_user): + from simple_module_core.events import EventBus + from users.contracts.events import UserRegistered + + bus = EventBus() + received: list[UserRegistered] = [] + + async def _on_registered(event: UserRegistered) -> None: + received.append(event) + + bus.subscribe(UserRegistered, _on_registered) + + request = MagicMock() + request.app.state.sm.event_bus = bus + + # Pre-verified so the manager doesn't try to send a verify email. + fake_user.is_verified = True + + await manager.on_after_register(fake_user, request) + + assert len(received) == 1 + assert received[0].user_id == fake_user.id + assert received[0].email == fake_user.email + + @pytest.mark.anyio + async def test_without_request_skips_publish(self, manager, fake_user): + """Bootstrap/CLI flows pass request=None — no bus means no publish, + and definitely no crash.""" + fake_user.is_verified = True + await manager.on_after_register(fake_user, None) + + class TestOnAfterForgotPassword: @pytest.mark.anyio async def test_calls_mailer_send_password_reset(self, manager, fake_mailer, fake_user): diff --git a/modules/users/users/manager.py b/modules/users/users/manager.py index 715c1ecd..213f6e39 100644 --- a/modules/users/users/manager.py +++ b/modules/users/users/manager.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import logging import uuid from datetime import UTC, datetime from typing import TYPE_CHECKING @@ -11,6 +12,7 @@ from fastapi_users import BaseUserManager, UUIDIDMixin, exceptions from fastapi_users.jwt import generate_jwt +from users.contracts.events import UserRegistered from users.db_adapter import UserDatabaseWithRoles, get_user_db from users.mailer import Mailer from users.models import User @@ -18,6 +20,8 @@ if TYPE_CHECKING: from users.settings import UsersSettings +logger = logging.getLogger(__name__) + class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]): """Customizes password validation, token secrets, and lifecycle emails.""" @@ -52,10 +56,26 @@ async def validate_password(self, password: str, user) -> None: # ── Lifecycle hooks ────────────────────────────────────── async def on_after_register(self, user: User, request: Request | None = None) -> None: + await self._publish_user_registered(user, request) if not user.is_verified: # Kicks off on_after_request_verify, which sends the email await self.request_verify(user, request) + async def _publish_user_registered(self, user: User, request: Request | None) -> None: + """Emit ``UserRegistered`` on the app-wide event bus. + + The bus lives on ``request.app.state.sm`` — unavailable when the + manager is exercised without a request (e.g. CLI bootstrap), so + we skip publication in that case rather than invent a bus. + """ + if request is None: + return + bus = getattr(getattr(request.app.state, "sm", None), "event_bus", None) + if bus is None: + logger.debug("Event bus missing from app.state.sm; skipping UserRegistered") + return + await bus.publish(UserRegistered(user_id=user.id, email=user.email)) + async def on_after_forgot_password( self, user: User, token: str, request: Request | None = None ) -> None: From 81ba7f8df34a5388fb1473992e47b404e3715f9e Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Apr 2026 18:44:01 +0000 Subject: [PATCH 3/3] Simplify event publish wiring and deterministic tests - _apply returns the upserted row so on_task_failure reads row.id directly instead of using a captured-list hack. - _publish_from_signal logs subscriber exceptions via done-callback rather than deferring until the future is GC'd. - Drop defensive getattr chain in _publish_user_registered: app.state.sm is a framework invariant. - Migrate TaskFailed/UserRegistered tests to FakeEventBus and replace a sleep-based wait with asyncio.Event + wait_for. --- .../background_tasks/signals.py | 38 ++++++++++++------- .../background_tasks/tests/test_signals.py | 27 ++++++------- modules/users/tests/test_user_manager.py | 20 +++------- modules/users/users/manager.py | 17 +++------ 4 files changed, 49 insertions(+), 53 deletions(-) diff --git a/modules/background_tasks/background_tasks/signals.py b/modules/background_tasks/background_tasks/signals.py index fbea53ce..fecedfb5 100644 --- a/modules/background_tasks/background_tasks/signals.py +++ b/modules/background_tasks/background_tasks/signals.py @@ -72,10 +72,22 @@ def _publish_from_signal(event: Any) -> None: if _bus is None or _loop is None: return try: - asyncio.run_coroutine_threadsafe(_bus.publish(event), _loop) + future = asyncio.run_coroutine_threadsafe(_bus.publish(event), _loop) except RuntimeError: - # Loop has stopped (shutdown race). Swallow — the DB row is already written. + # Loop has stopped (shutdown race). The DB row is already written. logger.debug("Event bus loop is not running; skipping %s", type(event).__name__) + return + # Surface subscriber exceptions — run_coroutine_threadsafe otherwise only + # logs them when the Future is GC'd, which happens far from the failure. + future.add_done_callback(_log_publish_failure) + + +def _log_publish_failure(future: asyncio.Future[Any]) -> None: + if future.cancelled(): + return + exc = future.exception() + if exc is not None: + logger.error("Event publish raised: %s", exc, exc_info=exc) def _apply( @@ -84,15 +96,21 @@ def _apply( celery_task_id: str | None, defaults: dict[str, Any], after: Callable[[TaskExecution], None] | None = None, -) -> None: - """Open a sync session, upsert by celery_task_id, log on failure.""" +) -> TaskExecution | None: + """Open a sync session, upsert by celery_task_id, log on failure. + + Returns the upserted row (or ``None`` if the session raised) so callers + can read DB-assigned fields like ``id`` without a second round trip. + """ try: with sync_session() as session: row = upsert_by_celery_id(session, celery_task_id=celery_task_id, defaults=defaults) if after is not None: after(row) + return row except Exception: logger.exception("%s failed for task_id=%s", handler, celery_task_id) + return None # ── Enqueue ───────────────────────────────────────────────────── @@ -207,12 +225,7 @@ def on_task_failure( task_name = task_name_of(sender) exception_type = type(exception).__name__ if exception is not None else None - published_id: list[Any] = [] - - def _capture_id(row: TaskExecution) -> None: - published_id.append(row.id) - - _apply( + row = _apply( "on_task_failure", celery_task_id=task_id, defaults={ @@ -222,13 +235,12 @@ def _capture_id(row: TaskExecution) -> None: "exception_type": exception_type, "finished_at": now_utc(), }, - after=_capture_id, ) - if published_id: + if row is not None: _publish_from_signal( TaskFailed( - task_execution_id=published_id[0], + task_execution_id=row.id, task_name=task_name, exception_type=exception_type, ) diff --git a/modules/background_tasks/tests/test_signals.py b/modules/background_tasks/tests/test_signals.py index 26d0a5ca..6f2c38a2 100644 --- a/modules/background_tasks/tests/test_signals.py +++ b/modules/background_tasks/tests/test_signals.py @@ -29,7 +29,7 @@ on_task_success, unbind_event_bus, ) -from simple_module_core.events import EventBus +from simple_module_testing import FakeEventBus from sqlalchemy import select @@ -199,15 +199,14 @@ class _Task: class TestTaskFailedEvent: @pytest.mark.asyncio async def test_publishes_when_bus_bound(self, sync_sqlite: Path): - bus = EventBus() - received: list[TaskFailed] = [] + bus = FakeEventBus() + dispatched = asyncio.Event() - async def _on_failed(event: TaskFailed) -> None: - received.append(event) + async def _mark(_event: TaskFailed) -> None: + dispatched.set() - bus.subscribe(TaskFailed, _on_failed) - loop = asyncio.get_running_loop() - bind_event_bus(bus, loop) + bus.subscribe(TaskFailed, _mark) + bind_event_bus(bus, asyncio.get_running_loop()) try: task_id = str(uuid.uuid4()) @@ -227,14 +226,12 @@ class _Einfo: sender=_Task, task_id=task_id, exception=ValueError("nope"), einfo=_Einfo() ) - # run_coroutine_threadsafe schedules the publish on this loop; - # yield long enough for the scheduled coroutine to complete. - await asyncio.sleep(0.05) + await asyncio.wait_for(dispatched.wait(), timeout=1.0) - assert len(received) == 1 - assert received[0].task_name == "demo.boom" - assert received[0].exception_type == "ValueError" - assert received[0].task_execution_id is not None + (event,) = bus.assert_published(TaskFailed) + assert event.task_name == "demo.boom" + assert event.exception_type == "ValueError" + assert event.task_execution_id is not None finally: unbind_event_bus() diff --git a/modules/users/tests/test_user_manager.py b/modules/users/tests/test_user_manager.py index 284fdf74..76521331 100644 --- a/modules/users/tests/test_user_manager.py +++ b/modules/users/tests/test_user_manager.py @@ -115,17 +115,10 @@ async def test_accepts_valid_password(self, manager, fake_user): class TestOnAfterRegister: @pytest.mark.anyio async def test_publishes_user_registered(self, manager, fake_user): - from simple_module_core.events import EventBus + from simple_module_testing import FakeEventBus from users.contracts.events import UserRegistered - bus = EventBus() - received: list[UserRegistered] = [] - - async def _on_registered(event: UserRegistered) -> None: - received.append(event) - - bus.subscribe(UserRegistered, _on_registered) - + bus = FakeEventBus() request = MagicMock() request.app.state.sm.event_bus = bus @@ -134,14 +127,13 @@ async def _on_registered(event: UserRegistered) -> None: await manager.on_after_register(fake_user, request) - assert len(received) == 1 - assert received[0].user_id == fake_user.id - assert received[0].email == fake_user.email + (event,) = bus.assert_published(UserRegistered) + assert event.user_id == fake_user.id + assert event.email == fake_user.email @pytest.mark.anyio async def test_without_request_skips_publish(self, manager, fake_user): - """Bootstrap/CLI flows pass request=None — no bus means no publish, - and definitely no crash.""" + """CLI/bootstrap flows pass request=None — must not crash.""" fake_user.is_verified = True await manager.on_after_register(fake_user, None) diff --git a/modules/users/users/manager.py b/modules/users/users/manager.py index 213f6e39..543f62df 100644 --- a/modules/users/users/manager.py +++ b/modules/users/users/manager.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio -import logging import uuid from datetime import UTC, datetime from typing import TYPE_CHECKING @@ -20,8 +19,6 @@ if TYPE_CHECKING: from users.settings import UsersSettings -logger = logging.getLogger(__name__) - class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]): """Customizes password validation, token secrets, and lifecycle emails.""" @@ -64,17 +61,15 @@ async def on_after_register(self, user: User, request: Request | None = None) -> async def _publish_user_registered(self, user: User, request: Request | None) -> None: """Emit ``UserRegistered`` on the app-wide event bus. - The bus lives on ``request.app.state.sm`` — unavailable when the - manager is exercised without a request (e.g. CLI bootstrap), so - we skip publication in that case rather than invent a bus. + CLI bootstrap and unit tests instantiate the manager without a + request, so there's no app context from which to reach the bus — + publication is best-effort in those cases. """ if request is None: return - bus = getattr(getattr(request.app.state, "sm", None), "event_bus", None) - if bus is None: - logger.debug("Event bus missing from app.state.sm; skipping UserRegistered") - return - await bus.publish(UserRegistered(user_id=user.id, email=user.email)) + await request.app.state.sm.event_bus.publish( + UserRegistered(user_id=user.id, email=user.email) + ) async def on_after_forgot_password( self, user: User, token: str, request: Request | None = None