From b9013c53fefde9ebbab9cc86e803242adbcc748d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Contreras=20Guill=C3=A9n?= Date: Thu, 7 May 2026 23:00:17 +0200 Subject: [PATCH] fix: clear all mypy --strict and ruff lint failures introduced in v0.3.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The v0.3.0 parity push merged a large amount of new code that hadn't been checked against the project's CI gates (mypy --strict, ruff check, ruff format). This commit makes the entire tree pass all four gates again. ## mypy --strict (67 → 0) - **Unused `# type: ignore[import-not-found]`** on optional vendor imports: with `uv sync --all-extras` the imports DO resolve, so the bare ignore is flagged. Switched to `[import-not-found, unused-ignore]` so the comment is harmless either way. - **`OrchestrationEvents` Protocol vs `_BaseOrchestrationEvents`**: the base class declared `**_: Any` no-op signatures, which subclass overrides narrowed. Re-declared the base methods with the same explicit keyword-arg signatures as the Protocol so subclassing is now strictly Liskov-substitutable. - **`Returning Any from function declared to return X`** in `step_invoker._compute_backoff`, `scheduling._compute_next_delay`, `cache_adapter.find`, `sqlalchemy_adapter.{delete, cleanup}`, `rule_engine.evaluator._eval_condition`, `config_server/backend._parse_text`, `config_server/client.fetch`, `client/protocols/graphql_client.execute`, `ecm/adapters/docusign.cancel`: cast/coerce return values to the declared type. - **`StepRecord | None` attribute access** in `transactional.workflow.executor`: lift `ctx.get_step(step.id)` to a local and None-check before reading `.attempts`. - **`auth` tuple type incompatibility** in `config_server.client`: only build the `(username, password)` tuple when both are non-None. - **Module-typed `_otel_trace = None` reassignment** in `tracer.py` and the same pattern in `scheduling.py`: declare `_otel_trace: Any = None` / `_croniter: Any = None` upfront and rebind on successful import. - **`yaml`, `croniter`, `grpc` library stubs**: add `[import-untyped]` to the `# type: ignore` lists where stubs are not available upstream. - **Module-level `_: Protocol = Impl(...)` probe lines** in 4 persistence / event-store adapters: removed; structural Protocol conformance is enforced by callers. - **Reassignment of `handler` variable** between command and query loops in `admin.providers.cqrs_provider` (pre-existing, surfaced by the new strict run): rename to `cmd_handler` / `query_handler`. ## ruff check (45 → 0) - **SIM105 `try/except/pass`** in `outbox.stop`, `projection.stop`, `recovery.stop`, `tracer.span`, `azure_blob.delete`: replace with `contextlib.suppress(...)`. - **SIM102 nested `if` statements** in `cache_adapter.cleanup` and `redis_adapter.cleanup`: collapse into a single boolean expression. - **SIM108 if/else block** in `rule_engine.evaluator._read`: collapse to a ternary. - **F401 unused imports**: stripped from various modules by `ruff check --fix`. ## ruff format `ruff format` reformatted 38 files (line wrapping consistent with the rest of the repo). ## Tests 2732 tests still passing. Zero behaviour change. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/pyfly/admin/providers/cqrs_provider.py | 12 +- src/pyfly/client/protocols/graphql_client.py | 5 +- src/pyfly/client/protocols/grpc_client.py | 5 +- src/pyfly/client/protocols/soap_client.py | 4 +- .../client/protocols/websocket_client.py | 2 +- src/pyfly/config_server/backend.py | 16 +- src/pyfly/config_server/client.py | 6 +- src/pyfly/config_server/server.py | 11 +- src/pyfly/ecm/adapters/adobe_sign.py | 6 +- src/pyfly/ecm/adapters/aws_s3.py | 10 +- src/pyfly/ecm/adapters/azure_blob.py | 11 +- src/pyfly/ecm/adapters/docusign.py | 5 +- src/pyfly/ecm/adapters/local_filesystem.py | 1 - src/pyfly/ecm/adapters/logalty.py | 11 +- src/pyfly/ecm/services.py | 2 - src/pyfly/eventsourcing/aggregate.py | 4 +- src/pyfly/eventsourcing/outbox.py | 5 +- src/pyfly/eventsourcing/projection.py | 11 +- src/pyfly/eventsourcing/repository.py | 10 +- src/pyfly/eventsourcing/store.py | 34 ++-- src/pyfly/idp/adapters/aws_cognito.py | 10 +- src/pyfly/idp/adapters/azure_ad.py | 11 +- src/pyfly/idp/adapters/internal_db.py | 4 +- src/pyfly/idp/adapters/keycloak.py | 6 +- src/pyfly/notifications/providers/dummy.py | 4 +- src/pyfly/notifications/providers/firebase.py | 2 +- src/pyfly/notifications/providers/resend.py | 2 +- src/pyfly/notifications/providers/sendgrid.py | 2 +- src/pyfly/notifications/providers/smtp.py | 4 +- src/pyfly/notifications/providers/twilio.py | 2 +- src/pyfly/rule_engine/dsl.py | 2 +- src/pyfly/rule_engine/evaluator.py | 9 +- .../adapters/event_loop/winloop_adapter.py | 2 +- src/pyfly/transactional/auto_configuration.py | 16 +- src/pyfly/transactional/core/argument.py | 5 +- src/pyfly/transactional/core/context.py | 10 +- src/pyfly/transactional/core/events.py | 148 +++++++++++++----- src/pyfly/transactional/core/metrics.py | 8 +- src/pyfly/transactional/core/persistence.py | 10 +- src/pyfly/transactional/core/recovery.py | 9 +- src/pyfly/transactional/core/scheduling.py | 18 +-- src/pyfly/transactional/core/step_invoker.py | 6 +- src/pyfly/transactional/core/tracer.py | 16 +- src/pyfly/transactional/core/validator.py | 4 +- .../persistence/cache_adapter.py | 16 +- .../persistence/redis_adapter.py | 11 +- .../persistence/sqlalchemy_adapter.py | 24 ++- src/pyfly/transactional/rest/controllers.py | 4 +- src/pyfly/transactional/saga/builder.py | 4 +- .../transactional/workflow/annotations.py | 19 +-- src/pyfly/transactional/workflow/engine.py | 20 +-- src/pyfly/transactional/workflow/executor.py | 7 +- tests/client/test_protocols.py | 5 +- tests/eventsourcing/test_eventsourcing.py | 8 +- tests/transactional/core/test_persistence.py | 4 +- tests/transactional/core/test_step_invoker.py | 4 +- tests/transactional/core/test_topology.py | 4 +- 57 files changed, 270 insertions(+), 341 deletions(-) diff --git a/src/pyfly/admin/providers/cqrs_provider.py b/src/pyfly/admin/providers/cqrs_provider.py index 796a2f43..8cfc6a40 100644 --- a/src/pyfly/admin/providers/cqrs_provider.py +++ b/src/pyfly/admin/providers/cqrs_provider.py @@ -44,24 +44,24 @@ async def get_handlers(self) -> dict[str, Any]: if reg.instance is not None and isinstance(reg.instance, HandlerRegistry): registry = reg.instance for cmd_type in registry.get_registered_command_types(): - handler = registry.find_command_handler(cmd_type) + cmd_handler = registry.find_command_handler(cmd_type) handlers.append( { "message_type": f"{cmd_type.__module__}.{cmd_type.__qualname__}", "message_name": cmd_type.__name__, - "handler_type": f"{type(handler).__module__}.{type(handler).__qualname__}", - "handler_name": type(handler).__name__, + "handler_type": f"{type(cmd_handler).__module__}.{type(cmd_handler).__qualname__}", + "handler_name": type(cmd_handler).__name__, "kind": "command", } ) for query_type in registry.get_registered_query_types(): - handler = registry.find_query_handler(query_type) + query_handler = registry.find_query_handler(query_type) handlers.append( { "message_type": f"{query_type.__module__}.{query_type.__qualname__}", "message_name": query_type.__name__, - "handler_type": f"{type(handler).__module__}.{type(handler).__qualname__}", - "handler_name": type(handler).__name__, + "handler_type": f"{type(query_handler).__module__}.{type(query_handler).__qualname__}", + "handler_name": type(query_handler).__name__, "kind": "query", } ) diff --git a/src/pyfly/client/protocols/graphql_client.py b/src/pyfly/client/protocols/graphql_client.py index aaa7a584..42ed11ef 100644 --- a/src/pyfly/client/protocols/graphql_client.py +++ b/src/pyfly/client/protocols/graphql_client.py @@ -30,7 +30,7 @@ async def execute( operation_name: str | None = None, ) -> dict[str, Any]: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "GraphQLClient requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc @@ -46,7 +46,8 @@ async def execute( data = resp.json() if "errors" in data: raise RuntimeError(f"GraphQL errors: {data['errors']}") - return data.get("data", {}) + payload: dict[str, Any] = data.get("data") or {} + return payload @dataclass diff --git a/src/pyfly/client/protocols/grpc_client.py b/src/pyfly/client/protocols/grpc_client.py index 1c436227..e1706310 100644 --- a/src/pyfly/client/protocols/grpc_client.py +++ b/src/pyfly/client/protocols/grpc_client.py @@ -41,7 +41,10 @@ def channel(self) -> Any: msg = "GrpcClientBuilder requires a target" raise ValueError(msg) try: - from grpc import aio, ssl_channel_credentials # type: ignore[import-not-found] + from grpc import ( # type: ignore[import-not-found, import-untyped, unused-ignore] + aio, + ssl_channel_credentials, + ) except ImportError as exc: # noqa: BLE001 msg = "GrpcClientBuilder requires grpcio — `pip install grpcio`" raise ImportError(msg) from exc diff --git a/src/pyfly/client/protocols/soap_client.py b/src/pyfly/client/protocols/soap_client.py index 7fc29c34..1dea31ba 100644 --- a/src/pyfly/client/protocols/soap_client.py +++ b/src/pyfly/client/protocols/soap_client.py @@ -8,8 +8,6 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any - _ENVELOPE = ( '' @@ -43,7 +41,7 @@ def __init__( async def call(self, body_xml: str) -> str: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "SoapClient requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc diff --git a/src/pyfly/client/protocols/websocket_client.py b/src/pyfly/client/protocols/websocket_client.py index 240a2b37..53486a76 100644 --- a/src/pyfly/client/protocols/websocket_client.py +++ b/src/pyfly/client/protocols/websocket_client.py @@ -27,7 +27,7 @@ def __init__( async def connect(self) -> Any: try: - import websockets # type: ignore[import-not-found] + import websockets # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "WebSocketClient requires websockets — `pip install websockets`" raise ImportError(msg) from exc diff --git a/src/pyfly/config_server/backend.py b/src/pyfly/config_server/backend.py index a812aaef..897f5687 100644 --- a/src/pyfly/config_server/backend.py +++ b/src/pyfly/config_server/backend.py @@ -76,9 +76,7 @@ async def fetch(self, application: str, profile: str, label: str = "main") -> Co properties = await asyncio.get_event_loop().run_in_executor( None, _parse_text, text, candidate.suffix.lstrip(".") ) - return ConfigSource( - application=application, profile=profile, label=label, properties=properties - ) + return ConfigSource(application=application, profile=profile, label=label, properties=properties) return None async def save(self, source: ConfigSource) -> None: @@ -87,9 +85,7 @@ async def save(self, source: ConfigSource) -> None: target = self._root / source.label if source.label else self._root target.mkdir(parents=True, exist_ok=True) path = target / f"{source.application}-{source.profile}.json" - await asyncio.get_event_loop().run_in_executor( - None, path.write_text, json.dumps(source.properties, indent=2) - ) + await asyncio.get_event_loop().run_in_executor(None, path.write_text, json.dumps(source.properties, indent=2)) async def list(self) -> list[ConfigSource]: results: list[ConfigSource] = [] @@ -114,7 +110,9 @@ def _parse_text(text: str, fmt: str) -> dict[str, Any]: if fmt == "json": import json - return json.loads(text) - import yaml + result: dict[str, Any] = json.loads(text) + return result + import yaml # type: ignore[import-untyped] - return yaml.safe_load(text) or {} + parsed: dict[str, Any] | None = yaml.safe_load(text) + return parsed or {} diff --git a/src/pyfly/config_server/client.py b/src/pyfly/config_server/client.py index 50296bfd..d48e05a7 100644 --- a/src/pyfly/config_server/client.py +++ b/src/pyfly/config_server/client.py @@ -32,13 +32,15 @@ def __init__( async def fetch(self) -> dict[str, Any]: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "ConfigClient requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc path = f"{self._url}/{self._application}/{self._profile}/{self._label}" - auth = (self._username, self._password) if self._username else None + auth: tuple[str, str] | None = ( + (self._username, self._password) if self._username is not None and self._password is not None else None + ) async with httpx.AsyncClient(timeout=15.0) as client: resp = await client.get(path, auth=auth) if resp.status_code != 200: diff --git a/src/pyfly/config_server/server.py b/src/pyfly/config_server/server.py index 3cb65de5..c008d11a 100644 --- a/src/pyfly/config_server/server.py +++ b/src/pyfly/config_server/server.py @@ -17,9 +17,7 @@ class ConfigServer: def __init__(self, backend: ConfigBackend) -> None: self._backend = backend - async def fetch( - self, application: str, profile: str = "default", label: str = "main" - ) -> dict[str, Any] | None: + async def fetch(self, application: str, profile: str = "default", label: str = "main") -> dict[str, Any] | None: source = await self._backend.fetch(application, profile, label) if source is None: return None @@ -42,14 +40,11 @@ async def save( properties: dict[str, Any], label: str = "main", ) -> dict[str, Any]: - source = ConfigSource( - application=application, profile=profile, label=label, properties=properties - ) + source = ConfigSource(application=application, profile=profile, label=label, properties=properties) await self._backend.save(source) return {"saved": True} async def list(self) -> list[dict[str, Any]]: return [ - {"application": s.application, "profile": s.profile, "label": s.label} - for s in await self._backend.list() + {"application": s.application, "profile": s.profile, "label": s.label} for s in await self._backend.list() ] diff --git a/src/pyfly/ecm/adapters/adobe_sign.py b/src/pyfly/ecm/adapters/adobe_sign.py index af5f83e9..355a4e85 100644 --- a/src/pyfly/ecm/adapters/adobe_sign.py +++ b/src/pyfly/ecm/adapters/adobe_sign.py @@ -30,7 +30,7 @@ def __init__(self, *, api_base: str, access_token: str) -> None: async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "AdobeSignESignatureAdapter requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc @@ -61,9 +61,7 @@ async def send(self, request: SignatureRequest) -> ESignatureEnvelope: "state": "IN_PROCESS", "message": request.message, } - resp = await client.post( - f"{self._api_base}/agreements", json=payload, headers=self._headers - ) + resp = await client.post(f"{self._api_base}/agreements", json=payload, headers=self._headers) resp.raise_for_status() data = resp.json() return ESignatureEnvelope( diff --git a/src/pyfly/ecm/adapters/aws_s3.py b/src/pyfly/ecm/adapters/aws_s3.py index 45d6c650..821dbb6f 100644 --- a/src/pyfly/ecm/adapters/aws_s3.py +++ b/src/pyfly/ecm/adapters/aws_s3.py @@ -33,13 +33,11 @@ def _ensure_client(self) -> Any: if self._client is not None: return self._client try: - import boto3 # type: ignore[import-not-found] + import boto3 # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "AwsS3StorageAdapter requires boto3 — `pip install boto3`" raise ImportError(msg) from exc - self._client = ( - boto3.client("s3", region_name=self._region) if self._region else boto3.client("s3") - ) + self._client = boto3.client("s3", region_name=self._region) if self._region else boto3.client("s3") return self._client async def _run(self, fn: Any, /, *args: Any, **kwargs: Any) -> Any: @@ -73,9 +71,7 @@ async def download(self, document: Document, version: int | None = None) -> byte raise FileNotFoundError(msg) target_version = version or document.versions[-1].version client = self._ensure_client() - resp = await self._run( - client.get_object, Bucket=self._bucket, Key=self._key(document.id, target_version) - ) + resp = await self._run(client.get_object, Bucket=self._bucket, Key=self._key(document.id, target_version)) body = resp["Body"].read() if hasattr(resp["Body"], "read") else resp["Body"] return body if isinstance(body, bytes) else bytes(body) diff --git a/src/pyfly/ecm/adapters/azure_blob.py b/src/pyfly/ecm/adapters/azure_blob.py index 7fa32395..eedeaaaf 100644 --- a/src/pyfly/ecm/adapters/azure_blob.py +++ b/src/pyfly/ecm/adapters/azure_blob.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio +import contextlib import hashlib from typing import Any @@ -36,7 +37,7 @@ def _ensure_service(self) -> Any: if self._service is not None: return self._service try: - from azure.storage.blob import BlobServiceClient # type: ignore[import-not-found] + from azure.storage.blob import BlobServiceClient # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "AzureBlobStorageAdapter requires azure-storage-blob — `pip install azure-storage-blob`" raise ImportError(msg) from exc @@ -44,9 +45,7 @@ def _ensure_service(self) -> Any: self._service = BlobServiceClient.from_connection_string(self._connection_string) else: assert self._account_url is not None - self._service = BlobServiceClient( - account_url=self._account_url, credential=self._credential - ) + self._service = BlobServiceClient(account_url=self._account_url, credential=self._credential) return self._service async def _run(self, fn: Any, /, *args: Any, **kwargs: Any) -> Any: @@ -84,10 +83,8 @@ async def delete(self, document: Document, version: int | None = None) -> bool: if version is None: for v in document.versions: blob = service.get_blob_client(container=self._container, blob=self._key(document.id, v.version)) - try: + with contextlib.suppress(Exception): await self._run(blob.delete_blob) - except Exception: # noqa: BLE001 - pass return bool(document.versions) blob = service.get_blob_client(container=self._container, blob=self._key(document.id, version)) try: diff --git a/src/pyfly/ecm/adapters/docusign.py b/src/pyfly/ecm/adapters/docusign.py index cf200494..490eb9ef 100644 --- a/src/pyfly/ecm/adapters/docusign.py +++ b/src/pyfly/ecm/adapters/docusign.py @@ -4,7 +4,6 @@ from __future__ import annotations -import asyncio import logging from datetime import UTC, datetime from typing import Any @@ -42,7 +41,7 @@ def __init__( async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "DocuSignESignatureAdapter requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc @@ -122,7 +121,7 @@ async def cancel(self, envelope_id: str) -> bool: json={"status": "voided", "voidedReason": "cancelled by application"}, headers=self._headers, ) - return resp.status_code == 200 + return bool(resp.status_code == 200) def _map_status(value: str) -> ESignatureStatus: diff --git a/src/pyfly/ecm/adapters/local_filesystem.py b/src/pyfly/ecm/adapters/local_filesystem.py index 9572ed53..b538f5a5 100644 --- a/src/pyfly/ecm/adapters/local_filesystem.py +++ b/src/pyfly/ecm/adapters/local_filesystem.py @@ -7,7 +7,6 @@ import asyncio import hashlib import pathlib -from typing import Any from pyfly.ecm.models import Document, DocumentVersion diff --git a/src/pyfly/ecm/adapters/logalty.py b/src/pyfly/ecm/adapters/logalty.py index 3231270e..c075d6a4 100644 --- a/src/pyfly/ecm/adapters/logalty.py +++ b/src/pyfly/ecm/adapters/logalty.py @@ -30,7 +30,7 @@ def __init__(self, *, api_base: str, api_key: str) -> None: async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "LogaltyESignatureAdapter requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc @@ -50,14 +50,9 @@ async def send(self, request: SignatureRequest) -> ESignatureEnvelope: "documentId": request.document_id, "subject": request.subject, "message": request.message, - "signers": [ - {"name": r.name, "email": r.email, "role": r.role} - for r in request.recipients - ], + "signers": [{"name": r.name, "email": r.email, "role": r.role} for r in request.recipients], } - resp = await client.post( - f"{self._api_base}/envelopes", json=payload, headers=self._headers - ) + resp = await client.post(f"{self._api_base}/envelopes", json=payload, headers=self._headers) resp.raise_for_status() data = resp.json() return ESignatureEnvelope( diff --git a/src/pyfly/ecm/services.py b/src/pyfly/ecm/services.py index 4aa8e2bc..000c6949 100644 --- a/src/pyfly/ecm/services.py +++ b/src/pyfly/ecm/services.py @@ -4,13 +4,11 @@ from __future__ import annotations -import asyncio import hashlib from datetime import UTC, datetime from pyfly.ecm.models import ( Document, - DocumentVersion, ESignatureEnvelope, Folder, SignatureRequest, diff --git a/src/pyfly/eventsourcing/aggregate.py b/src/pyfly/eventsourcing/aggregate.py index 9c7975d3..9d6c1178 100644 --- a/src/pyfly/eventsourcing/aggregate.py +++ b/src/pyfly/eventsourcing/aggregate.py @@ -34,9 +34,7 @@ class AggregateRoot: id: str = "" version: int = 0 _pending_events: list[DomainEvent] = field(default_factory=list, init=False, repr=False) - _handlers: dict[str, Callable[[Any, Any], None]] = field( - default_factory=dict, init=False, repr=False - ) + _handlers: dict[str, Callable[[Any, Any], None]] = field(default_factory=dict, init=False, repr=False) def when( self, diff --git a/src/pyfly/eventsourcing/outbox.py b/src/pyfly/eventsourcing/outbox.py index e1c75673..0ffa45fe 100644 --- a/src/pyfly/eventsourcing/outbox.py +++ b/src/pyfly/eventsourcing/outbox.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio +import contextlib import logging import uuid from collections.abc import Awaitable, Callable @@ -81,10 +82,8 @@ async def stop(self) -> None: if self._task is None: return self._stop.set() - try: + with contextlib.suppress(asyncio.CancelledError): await self._task - except asyncio.CancelledError: - pass self._task = None async def pending(self) -> list[OutboxRecord]: diff --git a/src/pyfly/eventsourcing/projection.py b/src/pyfly/eventsourcing/projection.py index a704a77d..542a8e00 100644 --- a/src/pyfly/eventsourcing/projection.py +++ b/src/pyfly/eventsourcing/projection.py @@ -16,9 +16,10 @@ from __future__ import annotations import asyncio +import contextlib import logging from collections.abc import Awaitable, Callable -from typing import Any, Protocol, runtime_checkable +from typing import Protocol, runtime_checkable from pyfly.eventsourcing.event import StoredEventEnvelope from pyfly.eventsourcing.store import EventStore @@ -62,10 +63,8 @@ async def stop(self) -> None: if self._task is None: return self._stop.set() - try: + with contextlib.suppress(asyncio.CancelledError): await self._task - except asyncio.CancelledError: - pass self._task = None async def _loop(self) -> None: @@ -94,9 +93,7 @@ async def _loop(self) -> None: class FunctionProjection: """Quick projection wrapper around a single async callable.""" - def __init__( - self, name: str, handler: Callable[[StoredEventEnvelope], Awaitable[None]] - ) -> None: + def __init__(self, name: str, handler: Callable[[StoredEventEnvelope], Awaitable[None]]) -> None: self.name = name self._handler = handler diff --git a/src/pyfly/eventsourcing/repository.py b/src/pyfly/eventsourcing/repository.py index 26e8e218..f8d728e2 100644 --- a/src/pyfly/eventsourcing/repository.py +++ b/src/pyfly/eventsourcing/repository.py @@ -76,9 +76,7 @@ async def save(self, aggregate: A) -> None: for evt in pending ] expected = aggregate.version - len(pending) - await self._store.append( - aggregate.id, aggregate_type, envelopes, expected_version=expected - ) + await self._store.append(aggregate.id, aggregate_type, envelopes, expected_version=expected) aggregate.mark_committed() if self._snapshots is not None and aggregate.version % self._snapshot_interval == 0: @@ -104,11 +102,7 @@ def _envelope_to_event(envelope: StoredEventEnvelope) -> object: @staticmethod def _dehydrate(aggregate: AggregateRoot) -> dict[str, object]: - return { - k: v - for k, v in vars(aggregate).items() - if not k.startswith("_") and k not in {"id", "version"} - } + return {k: v for k, v in vars(aggregate).items() if not k.startswith("_") and k not in {"id", "version"}} @staticmethod def _restore(aggregate: AggregateRoot, snapshot: Snapshot) -> None: diff --git a/src/pyfly/eventsourcing/store.py b/src/pyfly/eventsourcing/store.py index 9ff65f11..343cef2f 100644 --- a/src/pyfly/eventsourcing/store.py +++ b/src/pyfly/eventsourcing/store.py @@ -38,13 +38,9 @@ async def append( expected_version: int, ) -> None: ... - async def load( - self, aggregate_id: str, *, after_sequence: int = 0 - ) -> list[StoredEventEnvelope]: ... + async def load(self, aggregate_id: str, *, after_sequence: int = 0) -> list[StoredEventEnvelope]: ... - async def stream_all( - self, *, after_event_id: str | None = None, limit: int = 100 - ) -> list[StoredEventEnvelope]: ... + async def stream_all(self, *, after_event_id: str | None = None, limit: int = 100) -> list[StoredEventEnvelope]: ... async def latest_version(self, aggregate_id: str) -> int: ... @@ -78,16 +74,12 @@ async def append( self._all.append(evt) self._by_aggregate[aggregate_id] = current - async def load( - self, aggregate_id: str, *, after_sequence: int = 0 - ) -> list[StoredEventEnvelope]: + async def load(self, aggregate_id: str, *, after_sequence: int = 0) -> list[StoredEventEnvelope]: async with self._lock: events = self._by_aggregate.get(aggregate_id, []) return [e for e in events if e.sequence > after_sequence] - async def stream_all( - self, *, after_event_id: str | None = None, limit: int = 100 - ) -> list[StoredEventEnvelope]: + async def stream_all(self, *, after_event_id: str | None = None, limit: int = 100) -> list[StoredEventEnvelope]: async with self._lock: if after_event_id is None: return list(self._all[:limit]) @@ -128,7 +120,7 @@ def __init__(self, engine: Any) -> None: self._engine = engine async def initialize(self) -> None: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] async with self._engine.begin() as conn: await conn.execute(text(self.DDL)) @@ -141,7 +133,7 @@ async def append( *, expected_version: int, ) -> None: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] latest = await self.latest_version(aggregate_id) if latest != expected_version: @@ -176,7 +168,7 @@ async def append( ) async def load(self, aggregate_id: str, *, after_sequence: int = 0) -> list[StoredEventEnvelope]: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] async with self._engine.connect() as conn: rows = ( @@ -191,10 +183,8 @@ async def load(self, aggregate_id: str, *, after_sequence: int = 0) -> list[Stor ).fetchall() return [StoredEventEnvelope.from_json(r[0]) for r in rows] - async def stream_all( - self, *, after_event_id: str | None = None, limit: int = 100 - ) -> list[StoredEventEnvelope]: - from sqlalchemy import text # type: ignore[import-not-found] + async def stream_all(self, *, after_event_id: str | None = None, limit: int = 100) -> list[StoredEventEnvelope]: + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] async with self._engine.connect() as conn: if after_event_id is None: @@ -220,7 +210,7 @@ async def stream_all( return [StoredEventEnvelope.from_json(r[0]) for r in rows] async def latest_version(self, aggregate_id: str) -> int: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] async with self._engine.connect() as conn: result = await conn.execute( @@ -228,7 +218,3 @@ async def latest_version(self, aggregate_id: str) -> int: {"aid": aggregate_id}, ) return int(result.scalar() or 0) - - -_: EventStore = InMemoryEventStore() -_: EventStore = SqlAlchemyEventStore(engine=None) # type: ignore[arg-type, assignment] diff --git a/src/pyfly/idp/adapters/aws_cognito.py b/src/pyfly/idp/adapters/aws_cognito.py index bc3646c0..c17017af 100644 --- a/src/pyfly/idp/adapters/aws_cognito.py +++ b/src/pyfly/idp/adapters/aws_cognito.py @@ -48,7 +48,7 @@ def _ensure_client(self) -> Any: if self._client is not None: return self._client try: - import boto3 # type: ignore[import-not-found] + import boto3 # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "AwsCognitoIdpAdapter requires boto3 — `pip install boto3`" raise ImportError(msg) from exc @@ -87,9 +87,7 @@ async def create_user(self, user: IdpUser, password: str) -> IdpUser: async def get_user(self, user_id: str) -> IdpUser | None: client = self._ensure_client() try: - data = await self._run( - client.admin_get_user, UserPoolId=self._user_pool_id, Username=user_id - ) + data = await self._run(client.admin_get_user, UserPoolId=self._user_pool_id, Username=user_id) except Exception: # noqa: BLE001 return None return _from_cognito(data) @@ -203,9 +201,7 @@ async def reset_password(self, user_id: str) -> str: import secrets new_password = secrets.token_urlsafe(16) - await self.change_password( - PasswordChangeRequest(user_id=user_id, old_password="", new_password=new_password) - ) + await self.change_password(PasswordChangeRequest(user_id=user_id, old_password="", new_password=new_password)) return new_password async def assign_role(self, user_id: str, role: str) -> bool: diff --git a/src/pyfly/idp/adapters/azure_ad.py b/src/pyfly/idp/adapters/azure_ad.py index 7e9b3b1f..96657aee 100644 --- a/src/pyfly/idp/adapters/azure_ad.py +++ b/src/pyfly/idp/adapters/azure_ad.py @@ -55,7 +55,7 @@ def _token_url(self) -> str: async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "AzureAdIdpAdapter requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc @@ -221,9 +221,7 @@ async def reset_password(self, user_id: str) -> str: import secrets new_password = secrets.token_urlsafe(16) - await self.change_password( - PasswordChangeRequest(user_id=user_id, old_password="", new_password=new_password) - ) + await self.change_password(PasswordChangeRequest(user_id=user_id, old_password="", new_password=new_password)) return new_password async def assign_role(self, user_id: str, role: str) -> bool: @@ -251,10 +249,7 @@ async def list_roles(self) -> list[IdpRole]: headers = await self._app_auth_header() resp = await client.get(f"{self._graph}/groups", headers=headers) resp.raise_for_status() - return [ - IdpRole(name=g["id"], description=g.get("displayName", "")) - for g in resp.json().get("value", []) - ] + return [IdpRole(name=g["id"], description=g.get("displayName", "")) for g in resp.json().get("value", [])] def _from_aad(data: dict[str, Any]) -> IdpUser: diff --git a/src/pyfly/idp/adapters/internal_db.py b/src/pyfly/idp/adapters/internal_db.py index 2b4543ad..50850e70 100644 --- a/src/pyfly/idp/adapters/internal_db.py +++ b/src/pyfly/idp/adapters/internal_db.py @@ -161,7 +161,7 @@ async def list_roles(self) -> list[IdpRole]: @staticmethod def _hash(password: str) -> bytes: try: - import bcrypt # type: ignore[import-not-found] + import bcrypt # type: ignore[import-not-found, unused-ignore] return bcrypt.hashpw(password.encode(), bcrypt.gensalt()) except Exception: # noqa: BLE001 @@ -174,7 +174,7 @@ def _hash(password: str) -> bytes: @staticmethod def _verify(password: str, hashed: bytes) -> bool: try: - import bcrypt # type: ignore[import-not-found] + import bcrypt # type: ignore[import-not-found, unused-ignore] return bool(bcrypt.checkpw(password.encode(), hashed)) except Exception: # noqa: BLE001 diff --git a/src/pyfly/idp/adapters/keycloak.py b/src/pyfly/idp/adapters/keycloak.py index 628d58e5..8ea63744 100644 --- a/src/pyfly/idp/adapters/keycloak.py +++ b/src/pyfly/idp/adapters/keycloak.py @@ -68,7 +68,7 @@ def _token_url(self) -> str: async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "KeycloakIdpAdapter requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc @@ -243,9 +243,7 @@ async def reset_password(self, user_id: str) -> str: import secrets new_password = secrets.token_urlsafe(16) - await self.change_password( - PasswordChangeRequest(user_id=user_id, old_password="", new_password=new_password) - ) + await self.change_password(PasswordChangeRequest(user_id=user_id, old_password="", new_password=new_password)) return new_password async def assign_role(self, user_id: str, role: str) -> bool: diff --git a/src/pyfly/notifications/providers/dummy.py b/src/pyfly/notifications/providers/dummy.py index 96fd7cde..bc102e86 100644 --- a/src/pyfly/notifications/providers/dummy.py +++ b/src/pyfly/notifications/providers/dummy.py @@ -26,9 +26,7 @@ def __init__(self) -> None: async def send(self, message: EmailMessage) -> NotificationResult: self.sent.append(message) _logger.info("[dummy email] to=%s subject=%s", message.to, message.subject) - return NotificationResult( - id=message.id, provider=self.name, status=EmailStatus.SENT, provider_id=message.id - ) + return NotificationResult(id=message.id, provider=self.name, status=EmailStatus.SENT, provider_id=message.id) class DummySmsProvider: diff --git a/src/pyfly/notifications/providers/firebase.py b/src/pyfly/notifications/providers/firebase.py index 27535468..87cdeb36 100644 --- a/src/pyfly/notifications/providers/firebase.py +++ b/src/pyfly/notifications/providers/firebase.py @@ -25,7 +25,7 @@ def __init__(self, *, project_id: str, access_token: str) -> None: async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "FirebasePushProvider requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc diff --git a/src/pyfly/notifications/providers/resend.py b/src/pyfly/notifications/providers/resend.py index 4e2a2abd..b16bb788 100644 --- a/src/pyfly/notifications/providers/resend.py +++ b/src/pyfly/notifications/providers/resend.py @@ -20,7 +20,7 @@ def __init__(self, api_key: str, *, api_base: str = "https://api.resend.com") -> async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "ResendEmailProvider requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc diff --git a/src/pyfly/notifications/providers/sendgrid.py b/src/pyfly/notifications/providers/sendgrid.py index b6c6f252..5a9d3c3f 100644 --- a/src/pyfly/notifications/providers/sendgrid.py +++ b/src/pyfly/notifications/providers/sendgrid.py @@ -21,7 +21,7 @@ def __init__(self, api_key: str, *, api_base: str = "https://api.sendgrid.com/v3 async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "SendGridEmailProvider requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc diff --git a/src/pyfly/notifications/providers/smtp.py b/src/pyfly/notifications/providers/smtp.py index 83e6377b..f3ab5f77 100644 --- a/src/pyfly/notifications/providers/smtp.py +++ b/src/pyfly/notifications/providers/smtp.py @@ -52,9 +52,7 @@ def _send_blocking(self, message: EmailMessage) -> NotificationResult: if self._username and self._password: server.login(self._username, self._password) server.send_message(msg) - return NotificationResult( - id=message.id, provider=self.name, status=EmailStatus.SENT - ) + return NotificationResult(id=message.id, provider=self.name, status=EmailStatus.SENT) except Exception as exc: # noqa: BLE001 return NotificationResult( id=message.id, diff --git a/src/pyfly/notifications/providers/twilio.py b/src/pyfly/notifications/providers/twilio.py index 065d7571..f4c918db 100644 --- a/src/pyfly/notifications/providers/twilio.py +++ b/src/pyfly/notifications/providers/twilio.py @@ -21,7 +21,7 @@ def __init__(self, account_sid: str, auth_token: str, *, from_number: str | None async def _client(self) -> Any: try: - import httpx # type: ignore[import-not-found] + import httpx # type: ignore[import-not-found, unused-ignore] except ImportError as exc: # noqa: BLE001 msg = "TwilioSmsProvider requires httpx — `pip install pyfly[client]`" raise ImportError(msg) from exc diff --git a/src/pyfly/rule_engine/dsl.py b/src/pyfly/rule_engine/dsl.py index 7785b4f5..9ffabb4c 100644 --- a/src/pyfly/rule_engine/dsl.py +++ b/src/pyfly/rule_engine/dsl.py @@ -107,6 +107,6 @@ def from_dict(data: dict[str, Any]) -> RuleSet: @staticmethod def from_yaml(text: str) -> RuleSet: - import yaml + import yaml # type: ignore[import-untyped] return RuleSetLoader.from_dict(yaml.safe_load(text)) diff --git a/src/pyfly/rule_engine/evaluator.py b/src/pyfly/rule_engine/evaluator.py index ae852a95..5ff19e7c 100644 --- a/src/pyfly/rule_engine/evaluator.py +++ b/src/pyfly/rule_engine/evaluator.py @@ -51,9 +51,9 @@ def _eval_condition(self, c: Condition | None, ctx: dict[str, Any]) -> bool: actual = self._read(c.field, ctx) if c.field else None expected = c.value if op == "eq": - return actual == expected + return bool(actual == expected) if op == "ne": - return actual != expected + return bool(actual != expected) if op == "gt": return actual is not None and actual > expected if op == "ge": @@ -95,10 +95,7 @@ def _execute_action(self, action: Action, ctx: dict[str, Any]) -> None: def _read(path: str, ctx: dict[str, Any]) -> Any: cur: Any = ctx for part in path.split("."): - if isinstance(cur, dict): - cur = cur.get(part) - else: - cur = getattr(cur, part, None) + cur = cur.get(part) if isinstance(cur, dict) else getattr(cur, part, None) if cur is None: return None return cur diff --git a/src/pyfly/server/adapters/event_loop/winloop_adapter.py b/src/pyfly/server/adapters/event_loop/winloop_adapter.py index 4f1d6fde..2f30eb56 100644 --- a/src/pyfly/server/adapters/event_loop/winloop_adapter.py +++ b/src/pyfly/server/adapters/event_loop/winloop_adapter.py @@ -21,7 +21,7 @@ class WinloopEventLoopAdapter: def install(self) -> None: """Install winloop as the default asyncio event loop policy.""" - import winloop # type: ignore[import-not-found] + import winloop # type: ignore[import-not-found, unused-ignore] winloop.install() diff --git a/src/pyfly/transactional/auto_configuration.py b/src/pyfly/transactional/auto_configuration.py index 43752124..923aec26 100644 --- a/src/pyfly/transactional/auto_configuration.py +++ b/src/pyfly/transactional/auto_configuration.py @@ -65,6 +65,10 @@ from pyfly.transactional.saga.persistence.recovery import SagaRecoveryService from pyfly.transactional.saga.registry.saga_registry import SagaRegistry +# Shared (legacy adapters kept for back-compat). +from pyfly.transactional.shared.observability.events import LoggerEventsAdapter +from pyfly.transactional.shared.persistence.memory import InMemoryPersistenceAdapter + # tcc/ from pyfly.transactional.tcc.config.properties import TccEngineProperties from pyfly.transactional.tcc.engine.argument_resolver import TccArgumentResolver @@ -85,10 +89,6 @@ from pyfly.transactional.workflow.signal_service import SignalService from pyfly.transactional.workflow.timer_service import TimerService -# Shared (legacy adapters kept for back-compat). -from pyfly.transactional.shared.observability.events import LoggerEventsAdapter -from pyfly.transactional.shared.persistence.memory import InMemoryPersistenceAdapter - _logger = logging.getLogger(__name__) @@ -341,15 +341,11 @@ def saga_recovery_service( ) @bean - def orchestration_health_indicator( - self, persistence: ExecutionPersistenceProvider - ) -> OrchestrationHealthIndicator: + def orchestration_health_indicator(self, persistence: ExecutionPersistenceProvider) -> OrchestrationHealthIndicator: return OrchestrationHealthIndicator(persistence=persistence) @bean - def orchestration_controller( - self, persistence: ExecutionPersistenceProvider - ) -> OrchestrationController: + def orchestration_controller(self, persistence: ExecutionPersistenceProvider) -> OrchestrationController: return OrchestrationController(persistence=persistence) @bean diff --git a/src/pyfly/transactional/core/argument.py b/src/pyfly/transactional/core/argument.py index 523c4104..0acf8825 100644 --- a/src/pyfly/transactional/core/argument.py +++ b/src/pyfly/transactional/core/argument.py @@ -30,7 +30,6 @@ from pyfly.transactional.core.context import ExecutionContext - # --- Annotation markers ------------------------------------------------------- @@ -196,9 +195,7 @@ def _plans_for(self, method: Callable[..., Any], *, skip_first: bool) -> list[Re sig = inspect.signature(method) plans: list[ResolvedParameter] = [] params = list(sig.parameters.values()) - if skip_first and params and params[0].name in {"self", "cls"}: - params = params[1:] - elif skip_first and params: + if skip_first and params and params[0].name in {"self", "cls"} or skip_first and params: params = params[1:] try: diff --git a/src/pyfly/transactional/core/context.py b/src/pyfly/transactional/core/context.py index 86d21c94..792d2a50 100644 --- a/src/pyfly/transactional/core/context.py +++ b/src/pyfly/transactional/core/context.py @@ -22,7 +22,7 @@ import asyncio import uuid -from dataclasses import dataclass, field +from dataclasses import dataclass from datetime import UTC, datetime from typing import Any @@ -155,9 +155,7 @@ async def record_step_compensated(self, step_id: str, result: Any, error: BaseEx rec = self._steps.setdefault(step_id, StepRecord()) rec.compensation_result = result rec.compensation_error = str(error) if error else None - rec.status = ( - StepStatus.COMPENSATION_FAILED if error else StepStatus.COMPENSATED - ) + rec.status = StepStatus.COMPENSATION_FAILED if error else StepStatus.COMPENSATED self._touch() def get_step(self, step_id: str) -> StepRecord | None: @@ -302,9 +300,7 @@ def from_dict(cls, data: dict[str, Any]) -> ExecutionContext: ctx.tcc_phase = TccPhase(data["tcc_phase"]) ctx.started_at = datetime.fromisoformat(data["started_at"]) ctx.updated_at = datetime.fromisoformat(data["updated_at"]) - ctx.completed_at = ( - datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None - ) + ctx.completed_at = datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None ctx.error = data.get("error") ctx._variables = dict(data.get("variables", {})) for sid, raw in data.get("steps", {}).items(): diff --git a/src/pyfly/transactional/core/events.py b/src/pyfly/transactional/core/events.py index 563d4d91..29085569 100644 --- a/src/pyfly/transactional/core/events.py +++ b/src/pyfly/transactional/core/events.py @@ -37,54 +37,126 @@ class OrchestrationEvents(Protocol): """ async def on_start(self, *, name: str, pattern: ExecutionPattern, correlation_id: str) -> None: ... - async def on_completed(self, *, name: str, pattern: ExecutionPattern, correlation_id: str, success: bool, duration_ms: float) -> None: ... + async def on_completed( + self, *, name: str, pattern: ExecutionPattern, correlation_id: str, success: bool, duration_ms: float + ) -> None: ... async def on_step_started(self, *, name: str, correlation_id: str, step_id: str) -> None: ... - async def on_step_success(self, *, name: str, correlation_id: str, step_id: str, attempts: int, latency_ms: float) -> None: ... - async def on_step_failed(self, *, name: str, correlation_id: str, step_id: str, error: BaseException, attempts: int, latency_ms: float) -> None: ... + async def on_step_success( + self, *, name: str, correlation_id: str, step_id: str, attempts: int, latency_ms: float + ) -> None: ... + async def on_step_failed( + self, *, name: str, correlation_id: str, step_id: str, error: BaseException, attempts: int, latency_ms: float + ) -> None: ... async def on_step_skipped(self, *, name: str, correlation_id: str, step_id: str) -> None: ... async def on_compensation_started(self, *, name: str, correlation_id: str) -> None: ... - async def on_step_compensated(self, *, name: str, correlation_id: str, step_id: str, error: BaseException | None) -> None: ... + async def on_step_compensated( + self, *, name: str, correlation_id: str, step_id: str, error: BaseException | None + ) -> None: ... async def on_phase_started(self, *, name: str, correlation_id: str, phase: TccPhase) -> None: ... - async def on_phase_completed(self, *, name: str, correlation_id: str, phase: TccPhase, duration_ms: float) -> None: ... - async def on_phase_failed(self, *, name: str, correlation_id: str, phase: TccPhase, error: BaseException) -> None: ... - async def on_participant_started(self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str) -> None: ... - async def on_participant_success(self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str) -> None: ... - async def on_participant_failed(self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str, error: BaseException) -> None: ... + async def on_phase_completed( + self, *, name: str, correlation_id: str, phase: TccPhase, duration_ms: float + ) -> None: ... + async def on_phase_failed( + self, *, name: str, correlation_id: str, phase: TccPhase, error: BaseException + ) -> None: ... + async def on_participant_started( + self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str + ) -> None: ... + async def on_participant_success( + self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str + ) -> None: ... + async def on_participant_failed( + self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str, error: BaseException + ) -> None: ... async def on_workflow_suspended(self, *, name: str, correlation_id: str, reason: str) -> None: ... async def on_workflow_resumed(self, *, name: str, correlation_id: str) -> None: ... async def on_signal_delivered(self, *, name: str, correlation_id: str, signal: str) -> None: ... async def on_timer_fired(self, *, name: str, correlation_id: str, timer_id: str) -> None: ... - async def on_child_workflow_started(self, *, parent: str, correlation_id: str, child_workflow: str, child_correlation: str) -> None: ... - async def on_child_workflow_completed(self, *, parent: str, correlation_id: str, child_workflow: str, success: bool) -> None: ... + async def on_child_workflow_started( + self, *, parent: str, correlation_id: str, child_workflow: str, child_correlation: str + ) -> None: ... + async def on_child_workflow_completed( + self, *, parent: str, correlation_id: str, child_workflow: str, success: bool + ) -> None: ... async def on_continue_as_new(self, *, name: str, correlation_id: str, new_correlation_id: str) -> None: ... - async def on_dead_lettered(self, *, name: str, correlation_id: str, step_id: str | None, error: BaseException) -> None: ... + async def on_dead_lettered( + self, *, name: str, correlation_id: str, step_id: str | None, error: BaseException + ) -> None: ... class _BaseOrchestrationEvents: - """No-op base class so concrete adapters only override what they care about.""" - - async def on_start(self, **_: Any) -> None: ... - async def on_completed(self, **_: Any) -> None: ... - async def on_step_started(self, **_: Any) -> None: ... - async def on_step_success(self, **_: Any) -> None: ... - async def on_step_failed(self, **_: Any) -> None: ... - async def on_step_skipped(self, **_: Any) -> None: ... - async def on_compensation_started(self, **_: Any) -> None: ... - async def on_step_compensated(self, **_: Any) -> None: ... - async def on_phase_started(self, **_: Any) -> None: ... - async def on_phase_completed(self, **_: Any) -> None: ... - async def on_phase_failed(self, **_: Any) -> None: ... - async def on_participant_started(self, **_: Any) -> None: ... - async def on_participant_success(self, **_: Any) -> None: ... - async def on_participant_failed(self, **_: Any) -> None: ... - async def on_workflow_suspended(self, **_: Any) -> None: ... - async def on_workflow_resumed(self, **_: Any) -> None: ... - async def on_signal_delivered(self, **_: Any) -> None: ... - async def on_timer_fired(self, **_: Any) -> None: ... - async def on_child_workflow_started(self, **_: Any) -> None: ... - async def on_child_workflow_completed(self, **_: Any) -> None: ... - async def on_continue_as_new(self, **_: Any) -> None: ... - async def on_dead_lettered(self, **_: Any) -> None: ... + """No-op base class so concrete adapters only override what they care about. + + Method signatures intentionally mirror :class:`OrchestrationEvents` exactly, + so subclasses that override individual hooks satisfy mypy's strict + Liskov-substitution checks. + """ + + async def on_start(self, *, name: str, pattern: ExecutionPattern, correlation_id: str) -> None: ... + async def on_completed( + self, + *, + name: str, + pattern: ExecutionPattern, + correlation_id: str, + success: bool, + duration_ms: float, + ) -> None: ... + async def on_step_started(self, *, name: str, correlation_id: str, step_id: str) -> None: ... + async def on_step_success( + self, *, name: str, correlation_id: str, step_id: str, attempts: int, latency_ms: float + ) -> None: ... + async def on_step_failed( + self, + *, + name: str, + correlation_id: str, + step_id: str, + error: BaseException, + attempts: int, + latency_ms: float, + ) -> None: ... + async def on_step_skipped(self, *, name: str, correlation_id: str, step_id: str) -> None: ... + async def on_compensation_started(self, *, name: str, correlation_id: str) -> None: ... + async def on_step_compensated( + self, *, name: str, correlation_id: str, step_id: str, error: BaseException | None + ) -> None: ... + async def on_phase_started(self, *, name: str, correlation_id: str, phase: TccPhase) -> None: ... + async def on_phase_completed( + self, *, name: str, correlation_id: str, phase: TccPhase, duration_ms: float + ) -> None: ... + async def on_phase_failed( + self, *, name: str, correlation_id: str, phase: TccPhase, error: BaseException + ) -> None: ... + async def on_participant_started( + self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str + ) -> None: ... + async def on_participant_success( + self, *, name: str, correlation_id: str, phase: TccPhase, participant_id: str + ) -> None: ... + async def on_participant_failed( + self, + *, + name: str, + correlation_id: str, + phase: TccPhase, + participant_id: str, + error: BaseException, + ) -> None: ... + async def on_workflow_suspended(self, *, name: str, correlation_id: str, reason: str) -> None: ... + async def on_workflow_resumed(self, *, name: str, correlation_id: str) -> None: ... + async def on_signal_delivered(self, *, name: str, correlation_id: str, signal: str) -> None: ... + async def on_timer_fired(self, *, name: str, correlation_id: str, timer_id: str) -> None: ... + async def on_child_workflow_started( + self, *, parent: str, correlation_id: str, child_workflow: str, child_correlation: str + ) -> None: ... + async def on_child_workflow_completed( + self, *, parent: str, correlation_id: str, child_workflow: str, success: bool + ) -> None: ... + async def on_continue_as_new(self, *, name: str, correlation_id: str, new_correlation_id: str) -> None: ... + async def on_dead_lettered( + self, *, name: str, correlation_id: str, step_id: str | None, error: BaseException + ) -> None: ... class LoggerOrchestrationEvents(_BaseOrchestrationEvents): @@ -118,9 +190,7 @@ async def on_step_compensated( if error is None: _logger.info("[%s] %s.%s compensated", correlation_id, name, step_id) else: - _logger.error( - "[%s] compensation for %s.%s FAILED: %s", correlation_id, name, step_id, error - ) + _logger.error("[%s] compensation for %s.%s FAILED: %s", correlation_id, name, step_id, error) async def on_dead_lettered( self, *, name: str, correlation_id: str, step_id: str | None, error: BaseException diff --git a/src/pyfly/transactional/core/metrics.py b/src/pyfly/transactional/core/metrics.py index 2f706d51..179f5e43 100644 --- a/src/pyfly/transactional/core/metrics.py +++ b/src/pyfly/transactional/core/metrics.py @@ -132,9 +132,7 @@ def snapshot(self) -> dict[str, object]: "duration_p50_ms": self.execution_duration.get(name, _Histogram()).p50, "duration_p95_ms": self.execution_duration.get(name, _Histogram()).p95, } - for name in set(self.executions_started) - | set(self.executions_completed) - | set(self.executions_failed) + for name in set(self.executions_started) | set(self.executions_completed) | set(self.executions_failed) }, "steps": { key: { @@ -144,9 +142,7 @@ def snapshot(self) -> dict[str, object]: "p50_ms": self.step_latency.get(key, _Histogram()).p50, "p95_ms": self.step_latency.get(key, _Histogram()).p95, } - for key in set(self.steps_started) - | set(self.steps_succeeded) - | set(self.steps_failed) + for key in set(self.steps_started) | set(self.steps_succeeded) | set(self.steps_failed) }, "compensations": dict(self.compensations), "compensation_failures": dict(self.compensation_failures), diff --git a/src/pyfly/transactional/core/persistence.py b/src/pyfly/transactional/core/persistence.py index 2592640d..e7995acf 100644 --- a/src/pyfly/transactional/core/persistence.py +++ b/src/pyfly/transactional/core/persistence.py @@ -88,9 +88,7 @@ def deserialize(raw: str) -> ExecutionState: status=ExecutionStatus(data["status"]), started_at=datetime.fromisoformat(data["started_at"]), updated_at=datetime.fromisoformat(data["updated_at"]), - completed_at=( - datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None - ), + completed_at=(datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None), payload=data["payload"], ) @@ -144,11 +142,7 @@ async def find_all( async def find_stale(self, before: datetime) -> list[ExecutionState]: async with self._lock: - return [ - s - for s in self._store.values() - if not s.status.is_terminal and s.updated_at < before - ] + return [s for s in self._store.values() if not s.status.is_terminal and s.updated_at < before] async def delete(self, correlation_id: str) -> bool: async with self._lock: diff --git a/src/pyfly/transactional/core/recovery.py b/src/pyfly/transactional/core/recovery.py index 42ac2b98..d575440b 100644 --- a/src/pyfly/transactional/core/recovery.py +++ b/src/pyfly/transactional/core/recovery.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio +import contextlib import logging from datetime import UTC, datetime, timedelta @@ -71,10 +72,8 @@ async def stop(self) -> None: if self._task is None: return self._stop_event.set() - try: + with contextlib.suppress(asyncio.CancelledError): await self._task - except asyncio.CancelledError: - pass self._task = None async def _loop(self) -> None: @@ -97,8 +96,6 @@ async def _loop(self) -> None: except Exception as exc: # noqa: BLE001 _logger.error("recovery scan failed: %s", exc) try: - await asyncio.wait_for( - self._stop_event.wait(), timeout=self._scan_interval.total_seconds() - ) + await asyncio.wait_for(self._stop_event.wait(), timeout=self._scan_interval.total_seconds()) except TimeoutError: continue diff --git a/src/pyfly/transactional/core/scheduling.py b/src/pyfly/transactional/core/scheduling.py index 2c7df89a..4d5376bb 100644 --- a/src/pyfly/transactional/core/scheduling.py +++ b/src/pyfly/transactional/core/scheduling.py @@ -21,18 +21,20 @@ import asyncio import logging -import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta +from typing import Any +_croniter: Any = None +_HAS_CRONITER = False try: - from croniter import croniter as _croniter # type: ignore[import-not-found] + from croniter import croniter as _croniter_imported # type: ignore[import-untyped, import-not-found, unused-ignore] + _croniter = _croniter_imported _HAS_CRONITER = True except Exception: # noqa: BLE001 - _croniter = None - _HAS_CRONITER = False + pass _logger = logging.getLogger(__name__) @@ -107,9 +109,7 @@ async def stop(self) -> None: async def _run_loop(self, task: ScheduledTask) -> None: if task.initial_delay_ms > 0: try: - await asyncio.wait_for( - self._stop_event.wait(), timeout=task.initial_delay_ms / 1000.0 - ) + await asyncio.wait_for(self._stop_event.wait(), timeout=task.initial_delay_ms / 1000.0) return except TimeoutError: pass @@ -144,8 +144,8 @@ def _compute_next_delay(task: ScheduledTask) -> float: if task.cron is not None and _HAS_CRONITER: base = datetime.now(UTC) iter_ = _croniter(task.cron, base) - next_dt = iter_.get_next(datetime) - return max(0.0, (next_dt - base).total_seconds()) + next_dt: datetime = iter_.get_next(datetime) + return max(0.0, float((next_dt - base).total_seconds())) # Cron requested but croniter missing — back off long enough that the # admin notices when checking logs. _logger.warning("croniter not installed; cron task '%s' is inactive", task.id) diff --git a/src/pyfly/transactional/core/step_invoker.py b/src/pyfly/transactional/core/step_invoker.py index d240337c..50a168da 100644 --- a/src/pyfly/transactional/core/step_invoker.py +++ b/src/pyfly/transactional/core/step_invoker.py @@ -135,16 +135,14 @@ async def runner() -> Any: def _compute_backoff(policy: RetryPolicy, attempt: int) -> float: if policy.backoff_ms <= 0: return 0.0 - base_ms = policy.backoff_ms * (2 ** (attempt - 1)) + base_ms: float = float(policy.backoff_ms * (2 ** (attempt - 1))) if policy.jitter and policy.jitter_factor > 0: jitter_range = base_ms * policy.jitter_factor base_ms += random.uniform(0, jitter_range) return base_ms / 1000.0 @staticmethod - async def _apply_set_variables( - method: Callable[..., Any], kwargs: dict[str, Any], ctx: ExecutionContext - ) -> None: + async def _apply_set_variables(method: Callable[..., Any], kwargs: dict[str, Any], ctx: ExecutionContext) -> None: try: type_hints = typing.get_type_hints(method, include_extras=True) except Exception: diff --git a/src/pyfly/transactional/core/tracer.py b/src/pyfly/transactional/core/tracer.py index 148ea5a3..e404e41b 100644 --- a/src/pyfly/transactional/core/tracer.py +++ b/src/pyfly/transactional/core/tracer.py @@ -22,13 +22,15 @@ import contextlib from typing import Any +_otel_trace: Any = None +_HAS_OTEL = False try: - from opentelemetry import trace as _otel_trace # type: ignore[import-not-found] + from opentelemetry import trace as _otel_trace_imported # type: ignore[import-not-found, unused-ignore] + _otel_trace = _otel_trace_imported _HAS_OTEL = True except Exception: # noqa: BLE001 - _otel_trace = None - _HAS_OTEL = False + pass class OrchestrationTracer: @@ -40,9 +42,7 @@ class OrchestrationTracer: def __init__(self, service_name: str = "pyfly.orchestration") -> None: self._service_name = service_name - self._tracer: Any = ( - _otel_trace.get_tracer(service_name) if _HAS_OTEL else None - ) + self._tracer: Any = _otel_trace.get_tracer(service_name) if _HAS_OTEL else None @contextlib.contextmanager def span(self, name: str, **attributes: Any) -> Any: @@ -52,10 +52,8 @@ def span(self, name: str, **attributes: Any) -> Any: return with self._tracer.start_as_current_span(name) as span: for k, v in attributes.items(): - try: + with contextlib.suppress(Exception): span.set_attribute(k, v) - except Exception: # noqa: BLE001 - pass yield span def is_enabled(self) -> bool: diff --git a/src/pyfly/transactional/core/validator.py b/src/pyfly/transactional/core/validator.py index 2477090f..c16333f5 100644 --- a/src/pyfly/transactional/core/validator.py +++ b/src/pyfly/transactional/core/validator.py @@ -73,9 +73,7 @@ def validate_dag( try: TopologyBuilder.build_layers(graph) except TopologyError as exc: - report.issues.append( - ValidationIssue(target=target, level=IssueLevel.ERROR, message=str(exc)) - ) + report.issues.append(ValidationIssue(target=target, level=IssueLevel.ERROR, message=str(exc))) return report def fail_if_needed(self, report: ValidationReport) -> None: diff --git a/src/pyfly/transactional/persistence/cache_adapter.py b/src/pyfly/transactional/persistence/cache_adapter.py index 9dbe537e..c1885356 100644 --- a/src/pyfly/transactional/persistence/cache_adapter.py +++ b/src/pyfly/transactional/persistence/cache_adapter.py @@ -23,7 +23,6 @@ from typing import Any from pyfly.transactional.core.persistence import ( - ExecutionPersistenceProvider, ExecutionState, ) @@ -44,7 +43,8 @@ async def save(self, state: ExecutionState) -> None: self._index.add(state.correlation_id) async def find(self, correlation_id: str) -> ExecutionState | None: - return await self._cache.get(self._key(correlation_id)) + result: ExecutionState | None = await self._cache.get(self._key(correlation_id)) + return result async def find_all(self, *, status: Any = None, pattern: Any = None) -> list[ExecutionState]: results: list[ExecutionState] = [] @@ -72,13 +72,13 @@ async def cleanup(self, older_than: timedelta) -> int: cutoff = datetime.now(UTC) - older_than count = 0 for s in await self.find_all(): - if s.status.is_terminal and (s.completed_at or s.updated_at) < cutoff: - if await self.delete(s.correlation_id): - count += 1 + if ( + s.status.is_terminal + and (s.completed_at or s.updated_at) < cutoff + and await self.delete(s.correlation_id) + ): + count += 1 return count async def is_healthy(self) -> bool: return True - - -_: ExecutionPersistenceProvider = CachePersistenceProvider(cache_adapter=None) # type: ignore[arg-type] diff --git a/src/pyfly/transactional/persistence/redis_adapter.py b/src/pyfly/transactional/persistence/redis_adapter.py index 0c2290a6..c6adca19 100644 --- a/src/pyfly/transactional/persistence/redis_adapter.py +++ b/src/pyfly/transactional/persistence/redis_adapter.py @@ -23,7 +23,6 @@ from typing import Any from pyfly.transactional.core.persistence import ( - ExecutionPersistenceProvider, ExecutionState, StateSerializer, ) @@ -99,9 +98,12 @@ async def cleanup(self, older_than: timedelta) -> int: all_states = await self.find_all() count = 0 for s in all_states: - if s.status.is_terminal and (s.completed_at or s.updated_at) < cutoff: - if await self.delete(s.correlation_id): - count += 1 + if ( + s.status.is_terminal + and (s.completed_at or s.updated_at) < cutoff + and await self.delete(s.correlation_id) + ): + count += 1 return count async def is_healthy(self) -> bool: @@ -112,4 +114,3 @@ async def is_healthy(self) -> bool: # Make sure the adapter satisfies the structural Protocol. -_: ExecutionPersistenceProvider = RedisPersistenceProvider(redis_client=None) # type: ignore[arg-type] diff --git a/src/pyfly/transactional/persistence/sqlalchemy_adapter.py b/src/pyfly/transactional/persistence/sqlalchemy_adapter.py index 76d05d4c..cf1e3980 100644 --- a/src/pyfly/transactional/persistence/sqlalchemy_adapter.py +++ b/src/pyfly/transactional/persistence/sqlalchemy_adapter.py @@ -37,7 +37,6 @@ from typing import Any from pyfly.transactional.core.persistence import ( - ExecutionPersistenceProvider, ExecutionState, StateSerializer, ) @@ -64,13 +63,13 @@ def __init__(self, engine: Any, *, table_name: str = "pyfly_orchestration_state" self._table = table_name async def initialize(self) -> None: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] async with self._engine.begin() as conn: await conn.execute(text(self.DDL)) async def save(self, state: ExecutionState) -> None: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] raw = StateSerializer.serialize(state) sql = text( @@ -101,7 +100,7 @@ async def save(self, state: ExecutionState) -> None: ) async def find(self, correlation_id: str) -> ExecutionState | None: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] sql = text(f"SELECT payload FROM {self._table} WHERE correlation_id = :cid") async with self._engine.connect() as conn: @@ -111,7 +110,7 @@ async def find(self, correlation_id: str) -> ExecutionState | None: return StateSerializer.deserialize(row[0]) async def find_all(self, *, status: Any = None, pattern: Any = None) -> list[ExecutionState]: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] clauses: list[str] = [] params: dict[str, Any] = {} @@ -128,7 +127,7 @@ async def find_all(self, *, status: Any = None, pattern: Any = None) -> list[Exe return [StateSerializer.deserialize(r[0]) for r in rows] async def find_stale(self, before: datetime) -> list[ExecutionState]: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] sql = text( f"""SELECT payload FROM {self._table} @@ -140,15 +139,15 @@ async def find_stale(self, before: datetime) -> list[ExecutionState]: return [StateSerializer.deserialize(r[0]) for r in rows] async def delete(self, correlation_id: str) -> bool: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] sql = text(f"DELETE FROM {self._table} WHERE correlation_id = :cid") async with self._engine.begin() as conn: result = await conn.execute(sql, {"cid": correlation_id}) - return result.rowcount > 0 + return bool(result.rowcount > 0) async def cleanup(self, older_than: timedelta) -> int: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] cutoff = datetime.now(UTC) - older_than sql = text( @@ -158,17 +157,14 @@ async def cleanup(self, older_than: timedelta) -> int: ) async with self._engine.begin() as conn: result = await conn.execute(sql, {"cutoff": cutoff}) - return result.rowcount + return int(result.rowcount) async def is_healthy(self) -> bool: try: - from sqlalchemy import text # type: ignore[import-not-found] + from sqlalchemy import text # type: ignore[import-not-found, unused-ignore] async with self._engine.connect() as conn: await conn.execute(text("SELECT 1")) return True except Exception: # noqa: BLE001 return False - - -_: ExecutionPersistenceProvider = SqlAlchemyPersistenceProvider(engine=None) # type: ignore[arg-type] diff --git a/src/pyfly/transactional/rest/controllers.py b/src/pyfly/transactional/rest/controllers.py index 50d54c35..50c5e7c8 100644 --- a/src/pyfly/transactional/rest/controllers.py +++ b/src/pyfly/transactional/rest/controllers.py @@ -83,9 +83,7 @@ class DeadLetterController: def __init__(self, dlq: DeadLetterService) -> None: self._dlq = dlq - async def list( - self, execution_name: str | None = None, correlation_id: str | None = None - ) -> list[dict[str, Any]]: + async def list(self, execution_name: str | None = None, correlation_id: str | None = None) -> list[dict[str, Any]]: entries = await self._dlq.list(execution_name=execution_name, correlation_id=correlation_id) return [_dlq_to_dict(e) for e in entries] diff --git a/src/pyfly/transactional/saga/builder.py b/src/pyfly/transactional/saga/builder.py index e80ee1a5..91d1d9f2 100644 --- a/src/pyfly/transactional/saga/builder.py +++ b/src/pyfly/transactional/saga/builder.py @@ -26,9 +26,7 @@ class SagaBuilder: """Build a :class:`SagaDefinition` without decorators.""" def __init__(self, name: str, *, layer_concurrency: int = 0) -> None: - self._definition = SagaDefinition( - name=name, bean=None, layer_concurrency=layer_concurrency - ) + self._definition = SagaDefinition(name=name, bean=None, layer_concurrency=layer_concurrency) def step( self, diff --git a/src/pyfly/transactional/workflow/annotations.py b/src/pyfly/transactional/workflow/annotations.py index 3b046eb4..cfe07544 100644 --- a/src/pyfly/transactional/workflow/annotations.py +++ b/src/pyfly/transactional/workflow/annotations.py @@ -26,17 +26,6 @@ from dataclasses import dataclass from typing import Any -from pyfly.transactional.core.argument import ( - CorrelationId, - FromStep, - Header, - Headers, - Input, - Required, - SetVariable, - Variable, - Variables, -) from pyfly.transactional.core.model import TriggerMode # Re-export argument annotations under workflow.annotations so example code @@ -259,9 +248,7 @@ def decorator(fn: Callable[..., Any]) -> Callable[..., Any]: return decorator -def wait_for_all( - *signals: str, timeout_ms: int = 0 -) -> Callable[[Callable[..., Any]], Callable[..., Any]]: +def wait_for_all(*signals: str, timeout_ms: int = 0) -> Callable[[Callable[..., Any]], Callable[..., Any]]: def decorator(fn: Callable[..., Any]) -> Callable[..., Any]: fn.__pyfly_workflow_wait_all__ = WaitForAll(signals=signals, timeout_ms=timeout_ms) # type: ignore[attr-defined] return fn @@ -269,9 +256,7 @@ def decorator(fn: Callable[..., Any]) -> Callable[..., Any]: return decorator -def wait_for_any( - *signals: str, timeout_ms: int = 0 -) -> Callable[[Callable[..., Any]], Callable[..., Any]]: +def wait_for_any(*signals: str, timeout_ms: int = 0) -> Callable[[Callable[..., Any]], Callable[..., Any]]: def decorator(fn: Callable[..., Any]) -> Callable[..., Any]: fn.__pyfly_workflow_wait_any__ = WaitForAny(signals=signals, timeout_ms=timeout_ms) # type: ignore[attr-defined] return fn diff --git a/src/pyfly/transactional/workflow/engine.py b/src/pyfly/transactional/workflow/engine.py index bf2e7237..edffc44a 100644 --- a/src/pyfly/transactional/workflow/engine.py +++ b/src/pyfly/transactional/workflow/engine.py @@ -104,9 +104,7 @@ async def deliver_signal(self, correlation_id: str, signal: str, payload: Any = async def query(self, correlation_id: str, query_name: str, *args: Any, **kwargs: Any) -> Any: return await self._queries.query(correlation_id, query_name, *args, **kwargs) - async def list_executions( - self, *, status: ExecutionStatus | None = None - ) -> list[ExecutionState]: + async def list_executions(self, *, status: ExecutionStatus | None = None) -> list[ExecutionState]: return await self._persistence.find_all(status=status, pattern=ExecutionPattern.WORKFLOW) async def get_execution(self, correlation_id: str) -> ExecutionState | None: @@ -114,12 +112,8 @@ async def get_execution(self, correlation_id: str) -> ExecutionState | None: # --- private -------------------------------------------------------- - async def _start_async( - self, definition: Any, input: Any - ) -> WorkflowResult: - ctx = ExecutionContext( - name=definition.id, pattern=ExecutionPattern.WORKFLOW, input=input - ) + async def _start_async(self, definition: Any, input: Any) -> WorkflowResult: + ctx = ExecutionContext(name=definition.id, pattern=ExecutionPattern.WORKFLOW, input=input) await ctx.set_status(ExecutionStatus.PENDING) await self._persistence.save(ExecutionState.from_context(ctx)) asyncio.create_task(self._run(definition, input, preset_ctx=ctx)) @@ -137,9 +131,7 @@ async def _run( *, preset_ctx: ExecutionContext | None = None, ) -> WorkflowResult: - ctx = preset_ctx or ExecutionContext( - name=definition.id, pattern=ExecutionPattern.WORKFLOW, input=input - ) + ctx = preset_ctx or ExecutionContext(name=definition.id, pattern=ExecutionPattern.WORKFLOW, input=input) started = time.perf_counter() await self._signals.register(ctx) await self._queries.register(definition, ctx) @@ -153,9 +145,7 @@ async def _run( error: BaseException | None = None try: if definition.timeout_ms > 0: - await asyncio.wait_for( - self._executor.execute(definition, ctx), timeout=definition.timeout_ms / 1000.0 - ) + await asyncio.wait_for(self._executor.execute(definition, ctx), timeout=definition.timeout_ms / 1000.0) else: await self._executor.execute(definition, ctx) await ctx.set_status(ExecutionStatus.COMPLETED) diff --git a/src/pyfly/transactional/workflow/executor.py b/src/pyfly/transactional/workflow/executor.py index 9c368043..194839e8 100644 --- a/src/pyfly/transactional/workflow/executor.py +++ b/src/pyfly/transactional/workflow/executor.py @@ -95,9 +95,7 @@ async def _run_step( name=definition.id, correlation_id=ctx.correlation_id, reason=f"timer:{step.id}" ) await self._timers.sleep_ms(step.wait_for_timer_ms) - await self._events.on_timer_fired( - name=definition.id, correlation_id=ctx.correlation_id, timer_id=step.id - ) + await self._events.on_timer_fired(name=definition.id, correlation_id=ctx.correlation_id, timer_id=step.id) await self._events.on_workflow_resumed(name=definition.id, correlation_id=ctx.correlation_id) if step.wait_for_signal: @@ -168,11 +166,12 @@ async def _run_step( retry_policy=step.to_retry_policy(), ) elapsed = (time.perf_counter() - started) * 1000.0 + step_record = ctx.get_step(step.id) await self._events.on_step_success( name=definition.id, correlation_id=ctx.correlation_id, step_id=step.id, - attempts=ctx.get_step(step.id).attempts if ctx.get_step(step.id) else 1, + attempts=step_record.attempts if step_record is not None else 1, latency_ms=elapsed, ) on_step_cb = definition.on_step_callbacks.get(step.id) diff --git a/tests/client/test_protocols.py b/tests/client/test_protocols.py index bcad524e..c4d0cd49 100644 --- a/tests/client/test_protocols.py +++ b/tests/client/test_protocols.py @@ -55,10 +55,7 @@ def test_grpc_builder_requires_target() -> None: def test_websocket_builder_assembles_client() -> None: client = ( - WebSocketClientBuilder() - .with_url("wss://example.com/ws") - .with_header("Origin", "https://example.com") - .build() + WebSocketClientBuilder().with_url("wss://example.com/ws").with_header("Origin", "https://example.com").build() ) assert isinstance(client, WebSocketClient) assert client._url == "wss://example.com/ws" diff --git a/tests/eventsourcing/test_eventsourcing.py b/tests/eventsourcing/test_eventsourcing.py index e2802602..cec603a0 100644 --- a/tests/eventsourcing/test_eventsourcing.py +++ b/tests/eventsourcing/test_eventsourcing.py @@ -82,9 +82,7 @@ class TestRepository: @pytest.mark.asyncio async def test_save_and_load_round_trip(self) -> None: store = InMemoryEventStore() - repo: EventSourcedRepository[Order] = EventSourcedRepository( - store=store, factory=Order - ) + repo: EventSourcedRepository[Order] = EventSourcedRepository(store=store, factory=Order) order = Order() order.id = "o-1" order.apply(OrderPlaced(order_id="o-1", amount=99)) @@ -144,9 +142,7 @@ async def test_projection_consumes_events(self) -> None: async def collect(evt: StoredEventEnvelope) -> None: seen.append(evt) - runner = ProjectionRunner( - FunctionProjection("test", collect), store, poll_interval_s=0.05 - ) + runner = ProjectionRunner(FunctionProjection("test", collect), store, poll_interval_s=0.05) await runner.start() await asyncio.sleep(0.2) await runner.stop() diff --git a/tests/transactional/core/test_persistence.py b/tests/transactional/core/test_persistence.py index 2c5f1092..c6900b4e 100644 --- a/tests/transactional/core/test_persistence.py +++ b/tests/transactional/core/test_persistence.py @@ -73,9 +73,7 @@ async def test_find_stale(self, provider: InMemoryPersistenceProvider) -> None: assert len(stale) == 1 @pytest.mark.asyncio - async def test_cleanup_removes_terminal_old_records( - self, provider: InMemoryPersistenceProvider - ) -> None: + async def test_cleanup_removes_terminal_old_records(self, provider: InMemoryPersistenceProvider) -> None: s = _state(status=ExecutionStatus.COMPLETED) old = datetime.now(UTC) - timedelta(days=10) s.updated_at = old diff --git a/tests/transactional/core/test_step_invoker.py b/tests/transactional/core/test_step_invoker.py index 60359bb2..d3309401 100644 --- a/tests/transactional/core/test_step_invoker.py +++ b/tests/transactional/core/test_step_invoker.py @@ -49,9 +49,7 @@ async def test_sync_method_succeeds(self) -> None: def step(payload: Annotated[dict, Input()]) -> int: return payload["x"] + 1 - result = await invoker.invoke( - bean=None, method=step, step_id="s", ctx=_ctx(), retry_policy=RetryPolicy() - ) + result = await invoker.invoke(bean=None, method=step, step_id="s", ctx=_ctx(), retry_policy=RetryPolicy()) assert result == 2 @pytest.mark.asyncio diff --git a/tests/transactional/core/test_topology.py b/tests/transactional/core/test_topology.py index b5625f32..d0538759 100644 --- a/tests/transactional/core/test_topology.py +++ b/tests/transactional/core/test_topology.py @@ -20,9 +20,7 @@ def test_linear_dag(self) -> None: assert layers == [["a"], ["b"], ["c"]] def test_diamond_dag(self) -> None: - layers = TopologyBuilder.build_layers( - {"a": [], "b": ["a"], "c": ["a"], "d": ["b", "c"]} - ) + layers = TopologyBuilder.build_layers({"a": [], "b": ["a"], "c": ["a"], "d": ["b", "c"]}) assert layers[0] == ["a"] assert sorted(layers[1]) == ["b", "c"] assert layers[2] == ["d"]