From c71b5af35a391229e2fc0b6aa83c085f77a6f2b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Przemys=C5=82aw=20G=C3=B3recki?= Date: Thu, 6 Jul 2023 12:43:09 +0200 Subject: [PATCH] update transaction middleware --- .github/workflows/pytest.yml | 5 +- poetry.lock | 14 +++- pyproject.toml | 1 + src/api/tests/test_bidding.py | 2 +- src/config/container.py | 23 +++++-- src/seedwork/application/__init__.py | 68 +++++++++++-------- .../tests/application/test_application.py | 4 +- 7 files changed, 76 insertions(+), 41 deletions(-) diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 0b50991..cc813a1 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -35,7 +35,10 @@ jobs: uses: actions/setup-python@v3 with: python-version: "3.10" - - uses: Gr1N/setup-poetry@v7 + - name: Install Poetry + run: | + pip install poetry + poetry self update - name: Install dependencies run: poetry install - name: Run all tests diff --git a/poetry.lock b/poetry.lock index e703d44..a8586df 100644 --- a/poetry.lock +++ b/poetry.lock @@ -787,10 +787,21 @@ platformdirs = ">=2.4,<3" docs = ["proselint (>=0.13)", "sphinx (>=5.3)", "sphinx-argparse (>=0.3.2)", "sphinx-rtd-theme (>=1)", "towncrier (>=22.8)"] testing = ["coverage (>=6.2)", "coverage-enable-subprocess (>=1)", "flaky (>=3.7)", "packaging (>=21.3)", "pytest (>=7.0.1)", "pytest-env (>=0.6.2)", "pytest-freezegun (>=0.4.2)", "pytest-mock (>=3.6.1)", "pytest-randomly (>=3.10.3)", "pytest-timeout (>=2.1)"] +[[package]] +name = "vulture" +version = "2.7" +description = "Find dead code" +category = "dev" +optional = false +python-versions = ">=3.6" + +[package.dependencies] +toml = "*" + [metadata] lock-version = "1.1" python-versions = "^3.10.0" -content-hash = "39a1dd9a1ca117451b105bca0f7a6e62f0c5fe222d6158bdb0195510c320668d" +content-hash = "6eccbbfbc60a040faae5b4d7aba3e3233e060f5dc94d14452fc7c5b4510e7d48" [metadata.files] alembic = [] @@ -959,3 +970,4 @@ uvicorn = [ {file = "uvicorn-0.14.0.tar.gz", hash = "sha256:45ad7dfaaa7d55cab4cd1e85e03f27e9d60bc067ddc59db52a2b0aeca8870292"}, ] virtualenv = [] +vulture = [] diff --git a/pyproject.toml b/pyproject.toml index 1463e64..468e3b4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ bcrypt = "^4.0.1" poethepoet = "^0.10.0" pytest-cov = "^2.12.1" mypy = "^0.910" +vulture = "^2.7" [build-system] requires = ["poetry-core>=1.0.0"] diff --git a/src/api/tests/test_bidding.py b/src/api/tests/test_bidding.py index 8e31efa..20dab1d 100644 --- a/src/api/tests/test_bidding.py +++ b/src/api/tests/test_bidding.py @@ -37,7 +37,7 @@ def setup_app_for_bidding_tests(app, listing_id, seller_id, bidder_id): ) ) app.execute_command(PublishListingDraftCommand(listing_id=listing_id)) - logger.info("setup complete") + logger.info("test setup complete") @pytest.mark.integration diff --git a/src/config/container.py b/src/config/container.py index 2e4f3a8..a44544a 100644 --- a/src/config/container.py +++ b/src/config/container.py @@ -108,25 +108,36 @@ def on_enter_transaction_context(ctx): db_session=session, correlation_id=correlation_id, logger=logger ) ctx.dependency_provider = IocProvider(transaction_container) - logger.info(f"session={id(session)} transaction started") + logger.info(f"{id(session)} transaction started") @application.on_exit_transaction_context def on_exit_transaction_context(ctx, exc_type, exc_val, exc_tb): session = ctx.dependency_provider.get_dependency("db_session") if exc_type: session.rollback() - logger.info(f"session={id(session)} rollback due to {exc_type}") + logger.info(f"{id(session)} rollback due to {exc_type}") else: session.commit() - logger.info(f"session={id(session)} committed") + logger.info(f"{id(session)} committed") session.close() - logger.info(f"session={id(session)} transaction ended ") + logger.info(f"{id(session)} transaction ended ") @application.transaction_middleware - def logging_middleware(ctx, next): + def logging_middleware(ctx, next, command=None, query=None, event=None): + if command: + prefix = "Executing" + task = command + elif query: + prefix = "Querying" + task = query + elif event: + prefix = "Handling" + task = event + task = command or query or event session = ctx.dependency_provider.get_dependency("db_session") + logger.info(f"{id(session)} {prefix} {task}") result = next() - logger.info(f"Task {ctx.task} executed successfully") + logger.info(f"{id(session)} {prefix} completed, result: {result}") return result return application diff --git a/src/seedwork/application/__init__.py b/src/seedwork/application/__init__.py index 38689cc..598cd09 100644 --- a/src/seedwork/application/__init__.py +++ b/src/seedwork/application/__init__.py @@ -109,6 +109,14 @@ def __exit__(self, exc_type, exc_val, exc_tb): """Should be used to commit/end a transaction""" self.app._on_exit_transaction_context(self, exc_type, exc_val, exc_tb) + def _wrap_with_middlewares( + self, handler_func, command=None, query=None, event=None + ): + p = handler_func + for middleware in self.app._transaction_middlewares: + p = partial(middleware, self, p, command, query, event) + return p + def execute_query(self, query): assert ( self.task is None @@ -119,13 +127,9 @@ def execute_query(self, query): handler_kwargs = self.dependency_provider.get_handler_kwargs( handler_func, **self.overrides ) - - # prepare query handler - p = partial(handler_func, query, **handler_kwargs) - - for middleware in self.app._transaction_middlewares: - p = partial(middleware, self, p) - result = p() + handler_func = partial(handler_func, query, **handler_kwargs) + wrapped_handler = self._wrap_with_middlewares(handler_func, query=query) + result = wrapped_handler() return result def execute_command(self, command): @@ -138,20 +142,11 @@ def execute_command(self, command): handler_kwargs = self.dependency_provider.get_handler_kwargs( handler_func, **self.overrides ) - - # prepare command handler - p = partial(handler_func, command, **handler_kwargs) - - # wrap command handler in middlewares - handler_kwargs = self.dependency_provider.get_handler_kwargs( - handler_func, **self.overrides - ) - p = partial(handler_func, command, **handler_kwargs) - for middleware in self.app._transaction_middlewares: - p = partial(middleware, self, p) + handler_func = partial(handler_func, command, **handler_kwargs) + wrapped_handler = self._wrap_with_middlewares(handler_func, command=command) # execute wrapped command handler - command_result = p() + command_result = wrapped_handler() self.next_commands = [] self.integration_events = [] @@ -159,18 +154,12 @@ def execute_command(self, command): while len(event_queue) > 0: event = event_queue.pop(0) if isinstance(event, IntegrationEvent): - self.integration_events.append(result) + self._process_integration_event(event) + elif isinstance(event, DomainEvent): - for event_handler in self.app.get_event_handlers(event): - handler_kwargs = self.dependency_provider.get_handler_kwargs( - event_handler, **self.overrides - ) - logger.info(f"handling event {event} with {event_handler}") - result = event_handler(event, **handler_kwargs) - if isinstance(result, Command): - self.next_commands.append(result) - elif isinstance(result, EventResult): - event_queue.extend(result.events) + new_command, new_events = self._process_domain_event(event) + self.next_commands.extend(new_command) + event_queue.extend(new_events) return CommandResult.success(payload=command_result.payload) @@ -181,6 +170,25 @@ def get_service(self, service_cls): def current_user(self): return self.dependency_provider.get_dependency("current_user") + def _process_integration_event(self, event): + self.integration_events.append(event) + + def _process_domain_event(self, event): + new_commands = [] + new_events = [] + for handler_func in self.app.get_event_handlers(event): + handler_kwargs = self.dependency_provider.get_handler_kwargs( + handler_func, **self.overrides + ) + event_handler = partial(handler_func, event, **handler_kwargs) + wrapped_handler = self._wrap_with_middlewares(event_handler, event=event) + result = wrapped_handler() + if isinstance(result, Command): + new_commands.append(result) + elif isinstance(result, EventResult): + new_events.extend(result.events) + return new_commands, new_events + class ApplicationModule: def __init__(self, name, version=1.0): diff --git a/src/seedwork/tests/application/test_application.py b/src/seedwork/tests/application/test_application.py index c805d0e..3e5d429 100644 --- a/src/seedwork/tests/application/test_application.py +++ b/src/seedwork/tests/application/test_application.py @@ -124,12 +124,12 @@ def test_transaction_context_middleware(): app = Application(trace=[]) @app.transaction_middleware - def middleware1(ctx, next): + def middleware1(ctx, next, command=None, query=None, event=None): ctx.dependency_provider["trace"].append("middleware1") return next() @app.transaction_middleware - def middleware1(ctx, next): + def middleware1(ctx, next, command=None, query=None, event=None): ctx.dependency_provider["trace"].append("middleware2") return next()