diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index d0fb4c2..92bbd0d 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -23,7 +23,7 @@ jobs: - name: Create packages run: python -m build - name: Archive packages - uses: actions/upload-artifact@v3 + uses: actions/upload-artifact@v4 with: name: dist path: dist @@ -36,7 +36,7 @@ jobs: id-token: write steps: - name: Retrieve packages - uses: actions/download-artifact@v3 + uses: actions/download-artifact@v4 - name: Upload packages uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ab10332..85e99af 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -2,7 +2,7 @@ name: test suite on: push: - branches: [master] + branches: [master, "asphalt5"] pull_request: jobs: @@ -11,7 +11,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest] - python-version: ["3.8", "3.10", "3.11", "3.12", "pypy-3.9"] + python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"] runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v4 @@ -22,9 +22,9 @@ jobs: cache: pip cache-dependency-path: pyproject.toml - name: Start external services - run: docker-compose up -d + run: docker compose up -d - name: Install dependencies - run: pip install .[test] + run: pip install -e .[test] - name: Test with pytest run: coverage run -m pytest -v - name: Generate coverage report diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 45f073e..727687f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -5,7 +5,7 @@ # * Run "pre-commit install". repos: - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v4.6.0 + rev: v5.0.0 hooks: - id: check-toml - id: check-yaml @@ -16,17 +16,20 @@ repos: - id: trailing-whitespace - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.5.2 + rev: v0.8.4 hooks: - id: ruff args: [--fix, --show-fixes] - id: ruff-format - repo: https://github.com/pre-commit/mirrors-mypy - rev: v1.10.1 + rev: v1.14.0 hooks: - id: mypy - additional_dependencies: ["pytest"] + additional_dependencies: + - asphalt@git+https://github.com/asphalt-framework/asphalt + - pytest + - sqlalchemy - repo: https://github.com/pre-commit/pygrep-hooks rev: v1.10.0 diff --git a/.readthedocs.yml b/.readthedocs.yml index 026b967..ac1099a 100644 --- a/.readthedocs.yml +++ b/.readthedocs.yml @@ -3,7 +3,7 @@ version: 2 build: os: ubuntu-22.04 tools: - python: "3.8" + python: "3.9" sphinx: configuration: docs/conf.py diff --git a/docs/api.rst b/docs/api.rst index cc33c4e..65b5a0e 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -1,17 +1,15 @@ API reference ============= -Components ----------- +.. py:currentmodule:: asphalt.sqlalchemy -.. py:currentmodule:: asphalt.sqlalchemy.component +Component +--------- .. autoclass:: SQLAlchemyComponent -Utilities ---------- - -.. py:currentmodule:: asphalt.sqlalchemy.utils +Utility functions +----------------- .. autofunction:: clear_database .. autofunction:: clear_async_database diff --git a/docs/conf.py b/docs/conf.py index 8a356fb..22b23f5 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -7,6 +7,7 @@ "sphinx.ext.autodoc", "sphinx.ext.intersphinx", "sphinx_autodoc_typehints", + "sphinx_rtd_theme", "sphinx_tabs.tabs", ] @@ -25,6 +26,8 @@ exclude_patterns = ["_build"] pygments_style = "sphinx" +autodoc_default_options = {"members": True, "show-inheritance": True} +autodoc_inherit_docstrings = False highlight_language = "python3" todo_include_todos = False diff --git a/docs/configuration.rst b/docs/configuration.rst index 3773a02..a0da32f 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -5,16 +5,16 @@ Configuration A typical SQLAlchemy configuration consists of a single database. At minimum, you only need a connection URL (see the -`SQLAlchemy documentation`_ for how to construct one). Such a configuration would look -something like this:: +:doc:`SQLAlchemy documentation ` on how to construct one). +Such a configuration would look something like this:: components: sqlalchemy: - url: postgresql+asyncpg://user:password@10.0.0.8/mydatabase + url: postgresql+psycopg://user:password@10.0.0.8/mydatabase This will add two static resources and one resource factory, as follows: -* engine (type: :class:`sqlalchemy.future.engine.Engine` or +* engine (type: :class:`sqlalchemy.engine.Engine` or :class:`sqlalchemy.ext.asyncio.AsyncEngine`) * sessionmaker (type: :class:`sqlalchemy.orm.sessionmaker`) * session factory (generates :class:`~sqlalchemy.orm.Session` or @@ -26,7 +26,7 @@ sharing features):: components: sqlalchemy: url: - drivername: postgresql+asyncpg + drivername: postgresql+psycopg username: user password: password host: 10.0.0.8 @@ -34,15 +34,14 @@ sharing features):: .. seealso:: * :class:`sqlalchemy.engine.URL` - * :class:`asphalt.sqlalchemy.component.SQLAlchemyComponent` - -.. _SQLAlchemy documentation: https://docs.sqlalchemy.org/en/14/core/engines.html + * :class:`asphalt.sqlalchemy.SQLAlchemyComponent` Setting engine or session options --------------------------------- If you need to adjust the options used for creating new sessions, or pass extra -arguments to the engine, you can do so by specifying them in the ``session`` option:: +arguments to the engine, you can do so by specifying them in the ``session_args`` and +``engine_args``, respectively:: components: sqlalchemy: @@ -52,21 +51,3 @@ arguments to the engine, you can do so by specifying them in the ``session`` opt session_args: info: hello: world - -Multiple databases ------------------- - -If you need to work with multiple databases, you will need to use multiple instances -of the ``sqlalchemy`` component:: - - components: - sqlalchemy: - resource_name: db1 - url: postgresql+asyncpg:///mydatabase - sqlalchemy2: - type: sqlalchemy - resource_name: db2 - url: sqlite+aiosqlite:///mydb.sqlite - -This will make the appropriate resources available using their respective namespaces -(``db1`` or ``db2`` instead of ``default``). diff --git a/docs/events.rst b/docs/events.rst index 35e79ca..8223534 100644 --- a/docs/events.rst +++ b/docs/events.rst @@ -33,7 +33,7 @@ a synchronous ``sessionmaker`` resource is provided by this component even for a engines. To add listeners, simply use this session maker as the target for the listeners:: - from asphalt.core import NoCurrentContext, get_resource, inject, resource + from asphalt.core import NoCurrentContext, get_resource_nowait, inject, resource from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy import event @@ -41,7 +41,7 @@ listeners:: def precommit_hook(session): try: - async_session = get_resource(AsyncSession) + async_session = get_resource_nowait(AsyncSession) except NoCurrentContext: return diff --git a/docs/usage.rst b/docs/usage.rst index 60d6013..a28a304 100644 --- a/docs/usage.rst +++ b/docs/usage.rst @@ -46,31 +46,15 @@ you can skip this section. For those having to use engines for which no async counterpart is available, read on. Synchronous database connections in an asynchronous application pose a problem because -running operations against a database will block the event loop, and thus needs to be +running operations against a database will block the event loop, and thus need to be wrapped in worker threads. Another thing to watch out for is lazy loading of relationships and deferred columns which triggers implicit queries when those attributes are accessed. -On Python 3.9 and above, you would do:: - - from asyncio import to_thread - - from asphalt.core import inject, resource - from sqlalchemy.orm import Session - from sqlalchemy.future import select - - from yourapp.model import Person - - - @inject - async def handler(*, dbsession: Session = resource()) -> None: - people = await to_thread(dbsession.scalars, select(Person)) - - -On earlier Python versions:: - - from asyncio import get_running_loop +Here's an example of how you could use worker threads to run a blocking database +operation:: + from anyio import to_thread from asphalt.core import inject, resource from sqlalchemy.orm import Session from sqlalchemy.future import select @@ -80,8 +64,7 @@ On earlier Python versions:: @inject async def handler(*, dbsession: Session = resource()) -> None: - loop = get_running_loop() - people = await loop.run_in_executor(None, dbsession.scalars, select(Person)) + people = await to_thread.run_sync(dbsession.scalars, select(Person)) Releasing database resources during a long operation ---------------------------------------------------- diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index fb0543d..fd03e66 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -3,6 +3,13 @@ Version history This library adheres to `Semantic Versioning 2.0 `_. +**UNRELEASED** + +- Dropped support for Python 3.8 +- **BACKWARD INCOMPATIBLE** Bumped minimum Asphalt version to 5.0 +- **BACKWARD INCOMPATIBLE** Dropped the ``resource_name`` option in favor of Asphalt's + native alternate resource name syntax + **5.1.0** (2024-01-16) - Dropped support for Python 3.7 diff --git a/examples/csvimport.py b/examples/csvimport.py index a0c10fb..e766b91 100644 --- a/examples/csvimport.py +++ b/examples/csvimport.py @@ -9,9 +9,9 @@ import logging from pathlib import Path +from anyio import to_thread from asphalt.core import ( CLIApplicationComponent, - Context, inject, resource, run_application, @@ -37,23 +37,21 @@ class CSVImporterComponent(CLIApplicationComponent): def __init__(self) -> None: super().__init__() self.csv_path = Path(__file__).with_name("people.csv") - - async def start(self, ctx: Context) -> None: - # Remove the db file if it exists - db_path = self.csv_path.with_name("people.db") - if db_path.exists(): - db_path.unlink() - + self.db_path = self.csv_path.with_name("people.db") self.add_component( "sqlalchemy", - url=f"sqlite:///{db_path}", + url=f"sqlite:///{self.db_path}", ready_callback=lambda bind, factory: metadata.create_all(bind), ) - await super().start(ctx) + + async def prepare(self) -> None: + # Remove the db file if it exists + if self.db_path.exists(): + self.db_path.unlink() @inject - async def run(self, ctx: Context, *, dbsession: Session = resource()) -> None: - async with ctx.threadpool(): + async def run(self, *, dbsession: Session = resource()) -> None: + def insert_rows_in_thread() -> int: num_rows = 0 with self.csv_path.open() as csvfile: reader = csv.reader(csvfile, delimiter="|") @@ -65,7 +63,10 @@ async def run(self, ctx: Context, *, dbsession: Session = resource()) -> None: ) ) - logger.info("Imported %d rows of data", num_rows) + return num_rows + + inserted_rows = await to_thread.run_sync(insert_rows_in_thread) + logger.info("Imported %d rows of data", inserted_rows) -run_application(CSVImporterComponent(), logging=logging.DEBUG) +run_application(CSVImporterComponent) diff --git a/examples/csvimport_orm.py b/examples/csvimport_orm.py index cf0f41c..7c5efad 100644 --- a/examples/csvimport_orm.py +++ b/examples/csvimport_orm.py @@ -9,9 +9,9 @@ import logging from pathlib import Path +from anyio import to_thread from asphalt.core import ( CLIApplicationComponent, - Context, inject, resource, run_application, @@ -39,25 +39,23 @@ class CSVImporterComponent(CLIApplicationComponent): def __init__(self) -> None: super().__init__() self.csv_path = Path(__file__).with_name("people.csv") - - async def start(self, ctx: Context) -> None: - # Remove the db file if it exists - db_path = self.csv_path.with_name("people.db") - if db_path.exists(): - db_path.unlink() - + self.db_path = self.csv_path.with_name("people.db") self.add_component( "sqlalchemy", - url=f"sqlite:///{db_path}", + url=f"sqlite:///{self.db_path}", ready_callback=lambda bind, factory: DeclarativeBase.metadata.create_all( bind ), ) - await super().start(ctx) + + async def prepare(self) -> None: + # Remove the db file if it exists + if self.db_path.exists(): + self.db_path.unlink() @inject - async def run(self, ctx: Context, *, dbsession: Session = resource()) -> None: - async with ctx.threadpool(): + async def run(self, *, dbsession: Session = resource()) -> None: + def insert_rows_in_thread() -> int: num_rows = 0 with self.csv_path.open() as csvfile: reader = csv.reader(csvfile, delimiter="|") @@ -70,8 +68,10 @@ async def run(self, ctx: Context, *, dbsession: Session = resource()) -> None: # Emit pending INSERTs (though this would happen when the context ends # anyway) dbsession.flush() + return num_rows - logger.info("Imported %d rows of data", num_rows) + inserted_rows = await to_thread.run_sync(insert_rows_in_thread) + logger.info("Imported %d rows of data", inserted_rows) -run_application(CSVImporterComponent(), logging=logging.DEBUG) +run_application(CSVImporterComponent) diff --git a/pyproject.toml b/pyproject.toml index aa3dbd7..361fe8e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,15 +19,15 @@ classifiers = [ "Topic :: Database", "Programming Language :: Python", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", ] -requires-python = ">=3.8" +requires-python = ">=3.9" dependencies = [ - "asphalt ~= 4.9", + "asphalt @ git+https://github.com/asphalt-framework/asphalt", "SQLAlchemy[asyncio] >= 2.0", ] dynamic = ["version"] @@ -38,9 +38,8 @@ Homepage = "https://github.com/asphalt-framework/asphalt-sqlalchemy" [project.optional-dependencies] test = [ "aiosqlite", - "anyio >= 4.2", + "anyio[trio] >= 4.2", "asyncmy; platform_python_implementation == 'CPython'", - "asyncpg; platform_python_implementation == 'CPython'", "coverage >= 7", "pymysql", "psycopg >= 3.1; platform_python_implementation == 'CPython'", @@ -55,33 +54,37 @@ doc = [ ] [project.entry-points."asphalt.components"] -sqlalchemy = "asphalt.sqlalchemy.component:SQLAlchemyComponent" +sqlalchemy = "asphalt.sqlalchemy:SQLAlchemyComponent" [tool.setuptools_scm] version_scheme = "post-release" local_scheme = "dirty-tag" -[tool.ruff] +[tool.ruff.lint] select = [ "ASYNC", # flake8-async - "E", "F", "W", # default Flake8 "G", # flake8-logging-format "I", # isort "ISC", # flake8-implicit-str-concat "PGH", # pygrep-hooks "RUF100", # unused noqa (yesqa) "UP", # pyupgrade + "W", # pycodestyle warnings ] -[tool.ruff.isort] +[tool.ruff.lint.isort] known-first-party = ["asphalt.sqlalchemy"] +known-third-party = ["asphalt.core"] [tool.pytest.ini_options] -addopts = "-rsx --tb=short" +addopts = ["-rsfE", "--tb=short"] testpaths = ["tests"] [tool.mypy] -python_version = "3.8" +python_version = "3.9" +strict = true +explicit_package_bases = true +mypy_path = ["src", "tests", "examples"] [tool.coverage.run] source = ["asphalt.sqlalchemy"] @@ -92,20 +95,18 @@ branch = true show_missing = true [tool.tox] -legacy_tox_ini = """ -[tox] -envlist = py38, py39, py310, py311, py312, pypy3 +env_list = ["py39", "py310", "py311", "py312", "py313"] skip_missing_interpreters = true -minversion = 4.0 -[testenv] -extras = test -commands = python -m pytest {posargs} -setenv = - MYSQL_URL = mysql+pymysql://root@localhost:33060/asphalttest - POSTGRESQL_URL = postgresql+psycopg://postgres:secret@localhost:54320/asphalttest - -[testenv:docs] -extras = doc -commands = sphinx-build -W docs build/sphinx +[tool.tox.env_run_base] +commands = [["python", "-m", "pytest", { replace = "posargs", extend = true }]] +package = "editable" +extras = ["test"] +setenv = """ + MYSQL_URL = mysql+pymysql://root@localhost:33060/asphalttest + POSTGRESQL_URL = postgresql+psycopg://postgres:secret@localhost:54320/asphalttest """ + +[tool.tox.env.docs] +commands = [["sphinx-build", "-W", "-n", "docs", "build/sphinx", { replace = "posargs", extend = true }]] +extras = ["doc"] diff --git a/src/asphalt/sqlalchemy/__init__.py b/src/asphalt/sqlalchemy/__init__.py index e69de29..9dbb27a 100644 --- a/src/asphalt/sqlalchemy/__init__.py +++ b/src/asphalt/sqlalchemy/__init__.py @@ -0,0 +1,13 @@ +from typing import Any + +from ._component import SQLAlchemyComponent as SQLAlchemyComponent +from ._utils import apply_sqlite_hacks as apply_sqlite_hacks +from ._utils import clear_async_database as clear_async_database +from ._utils import clear_database as clear_database + +# Re-export imports, so they look like they live directly in this package +key: str +value: Any +for key, value in list(locals().items()): + if getattr(value, "__module__", "").startswith(f"{__name__}."): + value.__module__ = __name__ diff --git a/src/asphalt/sqlalchemy/component.py b/src/asphalt/sqlalchemy/_component.py similarity index 62% rename from src/asphalt/sqlalchemy/component.py rename to src/asphalt/sqlalchemy/_component.py index 52d73e7..b877aa4 100644 --- a/src/asphalt/sqlalchemy/component.py +++ b/src/asphalt/sqlalchemy/_component.py @@ -1,22 +1,22 @@ from __future__ import annotations import logging -from asyncio import get_running_loop -from collections.abc import AsyncGenerator, Callable -from concurrent.futures import ThreadPoolExecutor -from contextvars import copy_context +import sys +from collections.abc import Callable from inspect import isawaitable from typing import Any, cast +from anyio import CapacityLimiter, to_thread from asphalt.core import ( Component, - Context, - context_teardown, + add_resource, + add_resource_factory, + add_teardown_callback, qualified_name, resolve_reference, ) -from asphalt.sqlalchemy.utils import apply_sqlite_hacks +from asphalt.sqlalchemy._utils import apply_sqlite_hacks from sqlalchemy.engine import Connection, Engine, create_engine from sqlalchemy.engine.url import URL, make_url from sqlalchemy.exc import InvalidRequestError @@ -30,8 +30,6 @@ from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import Pool -logger = logging.getLogger(__name__) - class SQLAlchemyComponent(Component): """ @@ -43,7 +41,7 @@ class SQLAlchemyComponent(Component): For synchronous engines, the following resources are provided: - * :class:`~sqlalchemy.future.engine.Engine` + * :class:`~sqlalchemy.engine.Engine` * :class:`~sqlalchemy.orm.session.sessionmaker` * :class:`~sqlalchemy.orm.session.Session` @@ -57,18 +55,17 @@ class SQLAlchemyComponent(Component): .. note:: The following options will always be set to fixed values in sessions: * ``expire_on_commit``: ``False`` - * ``future``: ``True`` :param url: the connection url passed to - :func:`~sqlalchemy.future.engine.create_engine` + :func:`~sqlalchemy.create_engine` (can also be a dictionary of :class:`~sqlalchemy.engine.url.URL` keyword arguments) :param bind: a connection or engine to use instead of creating a new engine :param prefer_async: if ``True``, try to create an async engine rather than a synchronous one, in cases like ``psycopg`` where the driver supports both :param engine_args: extra keyword arguments passed to - :func:`sqlalchemy.future.engine.create_engine` or - :func:`sqlalchemy.ext.asyncio.create_engine` + :func:`sqlalchemy.create_engine` or + :func:`sqlalchemy.ext.asyncio.create_async_engine` :param session_args: extra keyword arguments passed to :class:`~sqlalchemy.orm.session.Session` or :class:`~sqlalchemy.ext.asyncio.AsyncSession` @@ -77,17 +74,13 @@ class SQLAlchemyComponent(Component): :param ready_callback: a callable that is called right before the resources are added to the context (can be a coroutine function too) :param poolclass: the SQLAlchemy pool class (or a textual reference to one) to use; - passed to :func:`sqlalchemy.future.engine.create_engine` or - :func:`sqlalchemy.ext.asyncio.create_engine` - :param resource_name: name space for the database resources + passed to :func:`sqlalchemy.create_engine` or + :func:`sqlalchemy.ext.asyncio.create_async_engine` """ - commit_executor: ThreadPoolExecutor - engine: Engine | AsyncEngine + _engine: Engine | AsyncEngine _bind: Connection | Engine - _sessionmaker: sessionmaker _async_bind: AsyncConnection | AsyncEngine - _async_sessionmaker: async_sessionmaker def __init__( self, @@ -97,13 +90,14 @@ def __init__( prefer_async: bool = True, engine_args: dict[str, Any] | None = None, session_args: dict[str, Any] | None = None, - commit_executor_workers: int = 5, - ready_callback: Callable[[Engine, sessionmaker], Any] | str | None = None, + commit_executor_workers: int = 50, + ready_callback: Callable[[Engine, sessionmaker[Any]], Any] + | Callable[[AsyncEngine, async_sessionmaker[Any]], Any] + | str + | None = None, poolclass: str | type[Pool] | None = None, - resource_name: str = "default", ): - self.resource_name = resource_name - self.commit_executor_workers = commit_executor_workers + self.commit_thread_limiter = CapacityLimiter(commit_executor_workers) self.ready_callback = resolve_reference(ready_callback) engine_args = engine_args or {} session_args = session_args or {} @@ -112,16 +106,16 @@ def __init__( if bind: if isinstance(bind, Connection): self._bind = bind - self.engine = bind.engine + self._engine = bind.engine elif isinstance(bind, AsyncConnection): self._async_bind = bind - self.engine = bind.engine + self._engine = bind.engine elif isinstance(bind, Engine): - self.engine = self._bind = bind + self._engine = self._bind = bind elif isinstance(bind, AsyncEngine): - self.engine = self._async_bind = bind + self._engine = self._async_bind = bind else: - raise TypeError(f"Incompatible bind argument: {qualified_name(bind)}") + raise TypeError(f"incompatible bind argument: {qualified_name(bind)}") else: if isinstance(url, dict): url = URL.create(**url) @@ -140,30 +134,30 @@ def __init__( if isinstance(poolclass, str): poolclass = resolve_reference(poolclass) - pool_class = cast("type[Pool]", poolclass) + pool_class = cast(type[Pool], poolclass) if prefer_async: try: - self.engine = self._async_bind = create_async_engine( + self._engine = self._async_bind = create_async_engine( url, poolclass=pool_class, **engine_args ) except InvalidRequestError: - self.engine = self._bind = create_engine( + self._engine = self._bind = create_engine( url, poolclass=pool_class, **engine_args ) else: try: - self.engine = self._bind = create_engine( + self._engine = self._bind = create_engine( url, poolclass=pool_class, **engine_args ) except InvalidRequestError: - self.engine = self._async_bind = create_async_engine( + self._engine = self._async_bind = create_async_engine( url, poolclass=pool_class, **engine_args ) if url.get_dialect().name == "sqlite": - apply_sqlite_hacks(self.engine) + apply_sqlite_hacks(self._engine) - if isinstance(self.engine, AsyncEngine): + if isinstance(self._engine, AsyncEngine): # This is needed for listening to ORM events when async sessions are used self._sessionmaker = sessionmaker() self._async_sessionmaker = async_sessionmaker( @@ -174,31 +168,30 @@ def __init__( else: self._sessionmaker = sessionmaker(bind=self._bind, **session_args) - def create_session(self, ctx: Context) -> Session: - async def teardown_session(exception: BaseException | None) -> None: + def create_session(self) -> Session: + async def teardown_session() -> None: try: if session.in_transaction(): - context = copy_context() - if exception is None: - await get_running_loop().run_in_executor( - self.commit_executor, context.run, session.commit + if sys.exc_info()[1] is None: + await to_thread.run_sync( + session.commit, limiter=self.commit_thread_limiter ) else: - await get_running_loop().run_in_executor( - self.commit_executor, context.run, session.rollback + await to_thread.run_sync( + session.rollback, limiter=self.commit_thread_limiter ) finally: session.close() session = self._sessionmaker() - ctx.add_teardown_callback(teardown_session, pass_exception=True) + add_teardown_callback(teardown_session) return session - def create_async_session(self, ctx: Context) -> AsyncSession: - async def teardown_session(exception: BaseException | None) -> None: + def create_async_session(self) -> AsyncSession: + async def teardown_session() -> None: try: if session.in_transaction(): - if exception is None: + if sys.exc_info()[1] is None: await session.commit() else: await session.rollback() @@ -206,26 +199,39 @@ async def teardown_session(exception: BaseException | None) -> None: await session.close() session: AsyncSession = self._async_sessionmaker() - ctx.add_teardown_callback(teardown_session, pass_exception=True) + add_teardown_callback(teardown_session) return session - @context_teardown - async def start(self, ctx: Context) -> AsyncGenerator[None, Exception | None]: + async def start(self) -> None: bind: Connection | Engine | AsyncConnection | AsyncEngine - if isinstance(self.engine, AsyncEngine): + if isinstance(self._engine, AsyncEngine): if self.ready_callback: retval = self.ready_callback(self._async_bind, self._sessionmaker) if isawaitable(retval): await retval bind = self._async_bind - ctx.add_resource(self.engine, self.resource_name) - ctx.add_resource(self._sessionmaker, self.resource_name) - ctx.add_resource(self._async_sessionmaker, self.resource_name) - ctx.add_resource_factory( + if isinstance(bind, AsyncEngine): + teardown_callback = self._engine.dispose + else: + teardown_callback = None + + add_resource( + self._engine, + description="SQLAlchemy engine (asynchronous)", + teardown_callback=teardown_callback, + ) + add_resource( + self._sessionmaker, + description="SQLAlchemy session factory (synchronous)", + ) + add_resource( + self._async_sessionmaker, + description="SQLAlchemy session factory (asynchronous)", + ) + add_resource_factory( self.create_async_session, - [AsyncSession], - self.resource_name, + description="SQLAlchemy session (asynchronous)", ) else: if self.ready_callback: @@ -233,30 +239,22 @@ async def start(self, ctx: Context) -> AsyncGenerator[None, Exception | None]: if isawaitable(retval): await retval - self.commit_executor = ThreadPoolExecutor(self.commit_executor_workers) - ctx.add_teardown_callback(self.commit_executor.shutdown) - bind = self._bind - ctx.add_resource(self.engine, self.resource_name) - ctx.add_resource(self._sessionmaker, self.resource_name) - ctx.add_resource_factory( + if isinstance(bind, AsyncEngine): + teardown_callback = self._engine.dispose + else: + teardown_callback = None + + add_resource( + self._engine, + description="SQLAlchemy engine (synchronous)", + teardown_callback=teardown_callback, + ) + add_resource( + self._sessionmaker, + description="SQLAlchemy session factory (synchronous)", + ) + add_resource_factory( self.create_session, - [Session], - self.resource_name, + description="SQLAlchemy session (synchronous)", ) - - logger.info( - "Configured SQLAlchemy resources (%s; dialect=%s, driver=%s)", - self.resource_name, - bind.dialect.name, - bind.dialect.driver, - ) - - yield - - if isinstance(bind, Engine): - bind.dispose() - elif isinstance(bind, AsyncEngine): - await bind.dispose() - - logger.info("SQLAlchemy resources (%s) shut down", self.resource_name) diff --git a/src/asphalt/sqlalchemy/utils.py b/src/asphalt/sqlalchemy/_utils.py similarity index 90% rename from src/asphalt/sqlalchemy/utils.py rename to src/asphalt/sqlalchemy/_utils.py index 3ca5bba..a2a6b79 100644 --- a/src/asphalt/sqlalchemy/utils.py +++ b/src/asphalt/sqlalchemy/_utils.py @@ -3,11 +3,10 @@ import sqlite3 from collections.abc import Iterable -from sqlalchemy import event +from sqlalchemy import Connection, Engine, MetaData +from sqlalchemy.event import listen from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine -from sqlalchemy.future import Connection, Engine from sqlalchemy.pool import ConnectionPoolEntry -from sqlalchemy.sql.schema import MetaData def clear_database(engine: Engine | Connection, schemas: Iterable[str] = ()) -> None: @@ -47,7 +46,11 @@ async def clear_async_database( for schema in all_schemas: # Reflect the schema to get the list of the tables, views and constraints metadata = MetaData() - await connection.run_sync(metadata.reflect, schema=schema, views=True) + await connection.run_sync( + metadata.reflect, + schema=schema, + views=True, + ) metadatas.append(metadata) for metadata in metadatas: @@ -63,11 +66,11 @@ def apply_sqlite_hacks(engine: Engine | AsyncEngine) -> None: integration tests (the connection is passed to the component as the ``bind`` option). + :param engine: an engine using the sqlite dialect + .. seealso:: https://docs.sqlalchemy.org/en/14/dialects/sqlite.html\ #pysqlite-serializable - :param engine: an engine using the sqlite dialect - """ def do_connect( @@ -88,5 +91,5 @@ def do_begin(conn: Connection) -> None: ) sync_engine = engine.sync_engine if isinstance(engine, AsyncEngine) else engine - event.listen(sync_engine, "connect", do_connect) - event.listen(sync_engine, "begin", do_begin) + listen(sync_engine, "connect", do_connect) + listen(sync_engine, "begin", do_begin) diff --git a/tests/conftest.py b/tests/conftest.py index df8ffa4..54dffa2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,52 +11,63 @@ from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine from sqlalchemy.future import Engine, create_engine -from asphalt.sqlalchemy.utils import apply_sqlite_hacks +from asphalt.sqlalchemy import apply_sqlite_hacks -@pytest.fixture(scope="session") -def anyio_backend() -> str: - return "asyncio" +@pytest.fixture +def aiosqlite_memory_url(anyio_backend_name: str) -> str: + if anyio_backend_name != "asyncio": + pytest.skip("Async SQLAlchemy only works with asyncio for now") + return "sqlite+aiosqlite:///:memory:" -@pytest.fixture(scope="session") -def psycopg_url() -> str: # type: ignore[return] - pytest.importorskip("asyncmy", reason="asyncmy is not available") + +@pytest.fixture +def psycopg_url() -> str: + pytest.importorskip("psycopg", reason="psycopg is not available") try: return os.environ["POSTGRESQL_URL"] except KeyError: pytest.skip("POSTGRESQL_URL environment variable is not set") -@pytest.fixture(scope="session") -def mysql_url() -> str: # type: ignore[return] +@pytest.fixture +def psycopg_url_async(psycopg_url: str, anyio_backend_name: str) -> str: + if anyio_backend_name != "asyncio": + pytest.skip("Async SQLAlchemy only works with asyncio for now") + + return psycopg_url + + +@pytest.fixture +def mysql_url() -> str: try: return os.environ["MYSQL_URL"] except KeyError: pytest.skip("MYSQL_URL environment variable is not set") -@pytest.fixture(scope="session") -def asyncmy_url(mysql_url: str) -> str: +@pytest.fixture +def asyncmy_url(mysql_url: str, anyio_backend_name: str) -> str: pytest.importorskip("asyncmy", reason="asyncmy is not available") return mysql_url.replace("pymysql", "asyncmy") -@pytest.fixture(scope="session") +@pytest.fixture def pymysql_engine(mysql_url: str) -> Generator[Engine, Any, None]: engine = create_engine(mysql_url) yield engine engine.dispose() -@pytest.fixture(scope="session") +@pytest.fixture def psycopg_engine(psycopg_url: str) -> Generator[Engine, Any, None]: engine = create_engine(psycopg_url, echo=True) yield engine engine.dispose() -@pytest.fixture(scope="session") +@pytest.fixture def sqlite_memory_engine() -> Generator[Engine, Any, None]: engine = create_engine( "sqlite:///:memory:", connect_args=dict(check_same_thread=False) @@ -66,7 +77,7 @@ def sqlite_memory_engine() -> Generator[Engine, Any, None]: engine.dispose() -@pytest.fixture(scope="session") +@pytest.fixture def sqlite_file_engine( tmp_path_factory: TempPathFactory, ) -> Generator[Engine, Any, None]: @@ -81,18 +92,23 @@ def sqlite_file_engine( db_path.unlink() -@pytest.fixture(scope="session") -async def aiosqlite_memory_engine() -> AsyncGenerator[AsyncEngine, Any]: - engine = create_async_engine("sqlite+aiosqlite:///:memory:") +@pytest.fixture +async def aiosqlite_memory_engine( + aiosqlite_memory_url: str, +) -> AsyncGenerator[AsyncEngine, Any]: + engine = create_async_engine(aiosqlite_memory_url) apply_sqlite_hacks(engine) yield engine await engine.dispose() -@pytest.fixture(scope="session") +@pytest.fixture async def aiosqlite_file_engine( - tmp_path_factory: TempPathFactory, + tmp_path_factory: TempPathFactory, anyio_backend_name: str ) -> AsyncGenerator[AsyncEngine, Any]: + if anyio_backend_name != "asyncio": + pytest.skip("Async SQLAlchemy only works with asyncio for now") + db_path = tmp_path_factory.mktemp("asphalt-sqlalchemy") / "test.db" engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}") apply_sqlite_hacks(engine) @@ -102,15 +118,25 @@ async def aiosqlite_file_engine( db_path.unlink() -@pytest.fixture(scope="session") -async def psycopg_async_engine(psycopg_url: str) -> AsyncGenerator[AsyncEngine, Any]: +@pytest.fixture +async def psycopg_async_engine( + psycopg_url: str, anyio_backend_name: str +) -> AsyncGenerator[AsyncEngine, Any]: + if anyio_backend_name != "asyncio": + pytest.skip("Async SQLAlchemy only works with asyncio for now") + engine = create_async_engine(psycopg_url) yield engine await engine.dispose() -@pytest.fixture(scope="session") -async def asyncmy_engine(asyncmy_url: str) -> AsyncGenerator[AsyncEngine, Any]: +@pytest.fixture +async def asyncmy_engine( + asyncmy_url: str, anyio_backend_name: str +) -> AsyncGenerator[AsyncEngine, Any]: + if anyio_backend_name != "asyncio": + pytest.skip("Async SQLAlchemy only works with asyncio for now") + engine = create_async_engine(asyncmy_url) yield engine await engine.dispose() @@ -123,7 +149,6 @@ async def asyncmy_engine(asyncmy_url: str) -> AsyncGenerator[AsyncEngine, Any]: lf("pymysql_engine"), lf("psycopg_engine"), ], - scope="session", ) def sync_engine(request: SubRequest) -> Engine: return cast(Engine, request.param) @@ -136,7 +161,6 @@ def sync_engine(request: SubRequest) -> Engine: lf("psycopg_async_engine"), lf("asyncmy_engine"), ], - scope="session", ) async def async_engine(request: SubRequest) -> AsyncEngine: return cast(AsyncEngine, request.param) diff --git a/tests/test_component.py b/tests/test_component.py index 05a5722..e3b438f 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -1,14 +1,18 @@ from __future__ import annotations import gc -from contextlib import ExitStack +from contextlib import AsyncExitStack from pathlib import Path from threading import Thread, current_thread -from typing import Any +from typing import Any, cast import pytest -from asphalt.core import NoCurrentContext, current_context -from asphalt.core.context import Context, get_resource +from asphalt.core import ( + Context, + NoCurrentContext, + current_context, + get_resource_nowait, +) from pytest import FixtureRequest from sqlalchemy.engine.url import URL from sqlalchemy.event import listen, remove @@ -23,55 +27,41 @@ from sqlalchemy.pool import AsyncAdaptedQueuePool, NullPool, QueuePool from sqlalchemy.sql import text -from asphalt.sqlalchemy.component import SQLAlchemyComponent -from asphalt.sqlalchemy.utils import clear_async_database, clear_database +from asphalt.sqlalchemy import SQLAlchemyComponent, clear_async_database, clear_database from .model import Person pytestmark = pytest.mark.anyio -@pytest.mark.parametrize( - "component_opts, args", - [ - pytest.param({}, ()), - pytest.param({"resource_name": "alternate"}, ("alternate",)), - ], -) -async def test_component_start_sync( - component_opts: dict[str, Any], args: tuple[Any] -) -> None: +def test_bad_bind_argument() -> None: + with pytest.raises(TypeError, match="incompatible bind argument: str"): + SQLAlchemyComponent(bind="bad") # type: ignore[arg-type] + + +async def test_component_start_sync() -> None: """Test that the component creates all the expected (synchronous) resources.""" url = URL.create("sqlite", database=":memory:") - component = SQLAlchemyComponent(url=url, **component_opts) - async with Context() as ctx: - await component.start(ctx) + component = SQLAlchemyComponent(url=url) + async with Context(): + await component.start() - ctx.require_resource(Engine, *args) - ctx.require_resource(sessionmaker, *args) - ctx.require_resource(Session, *args) + get_resource_nowait(Engine) + get_resource_nowait(sessionmaker) + get_resource_nowait(Session) -@pytest.mark.parametrize( - "component_opts, args", - [ - pytest.param({}, ()), - pytest.param({"resource_name": "alternate"}, ("alternate",)), - ], -) -async def test_component_start_async( - component_opts: dict[str, Any], args: tuple[Any] -) -> None: +async def test_component_start_async() -> None: """Test that the component creates all the expected (asynchronous) resources.""" url = URL.create("sqlite+aiosqlite", database=":memory:") - component = SQLAlchemyComponent(url=url, **component_opts) - async with Context() as ctx: - await component.start(ctx) - - ctx.require_resource(AsyncEngine, *args) - async_session_class = ctx.require_resource(async_sessionmaker, *args) - ctx.require_resource(AsyncSession, *args) - sync_session_class = ctx.require_resource(sessionmaker, *args) + component = SQLAlchemyComponent(url=url) + async with Context(): + await component.start() + + get_resource_nowait(AsyncEngine) + async_session_class = get_resource_nowait(async_sessionmaker) + get_resource_nowait(AsyncSession) + sync_session_class = get_resource_nowait(sessionmaker) assert async_session_class.kw["sync_session_class"] is sync_session_class @@ -80,15 +70,15 @@ async def test_component_start_async( ) async def test_ready_callback(asynchronous: bool) -> None: engine2: Engine | AsyncEngine | None = None - factory2: sessionmaker | async_sessionmaker | None = None + factory2: sessionmaker[Any] | async_sessionmaker[Any] | None = None - def ready_callback(engine: Engine, factory: sessionmaker) -> None: + def ready_callback(engine: Engine, factory: sessionmaker[Any]) -> None: nonlocal engine2, factory2 engine2 = engine factory2 = factory async def ready_callback_async( - engine: AsyncEngine, factory: async_sessionmaker + engine: AsyncEngine, factory: async_sessionmaker[Any] ) -> None: nonlocal engine2, factory2 engine2 = engine @@ -96,11 +86,11 @@ async def ready_callback_async( callback = ready_callback_async if asynchronous else ready_callback component = SQLAlchemyComponent(url="sqlite:///:memory:", ready_callback=callback) - async with Context() as ctx: - await component.start(ctx) + async with Context(): + await component.start() - engine = ctx.require_resource(Engine) - factory = ctx.require_resource(sessionmaker) + engine = get_resource_nowait(Engine) + factory = get_resource_nowait(sessionmaker) assert engine is engine2 assert factory is factory2 @@ -110,25 +100,25 @@ async def test_bind_sync_connection() -> None: engine = create_engine("sqlite:///:memory:") connection = engine.connect() component = SQLAlchemyComponent(bind=connection) - async with Context() as ctx: - await component.start(ctx) + async with Context(): + await component.start() - assert ctx.require_resource(Engine) is engine - assert ctx.require_resource(Session).bind is connection + assert get_resource_nowait(Engine) is engine + assert get_resource_nowait(Session).bind is connection connection.close() -async def test_bind_async_connection() -> None: +async def test_bind_async_connection(aiosqlite_memory_url: str) -> None: """Test that a Connection can be passed as "bind" in place of "url".""" - engine = create_async_engine("sqlite+aiosqlite:///:memory:") + engine = create_async_engine(aiosqlite_memory_url) connection = await engine.connect() component = SQLAlchemyComponent(bind=connection) - async with Context() as ctx: - await component.start(ctx) + async with Context(): + await component.start() - assert ctx.require_resource(AsyncEngine) is engine - assert ctx.require_resource(AsyncSession).bind is connection + assert get_resource_nowait(AsyncEngine) is engine + assert get_resource_nowait(AsyncSession).bind is connection await connection.close() @@ -137,11 +127,11 @@ async def test_bind_sync_engine() -> None: """Test that a Connection can be passed as "bind" in place of "url".""" engine = create_engine("sqlite:///:memory:") component = SQLAlchemyComponent(bind=engine) - async with Context() as ctx: - await component.start(ctx) + async with Context(): + await component.start() - assert ctx.require_resource(Engine) is engine - assert ctx.require_resource(Session).bind is engine + assert get_resource_nowait(Engine) is engine + assert get_resource_nowait(Session).bind is engine engine.dispose() @@ -150,11 +140,11 @@ async def test_bind_async_engine() -> None: """Test that a Connection can be passed as "bind" in place of "url".""" engine = create_async_engine("sqlite+aiosqlite:///:memory:") component = SQLAlchemyComponent(bind=engine) - async with Context() as ctx: - await component.start(ctx) + async with Context(): + await component.start() - assert ctx.require_resource(AsyncEngine) is engine - assert ctx.require_resource(AsyncSession).bind is engine + assert get_resource_nowait(AsyncEngine) is engine + assert get_resource_nowait(AsyncSession).bind is engine await engine.dispose() @@ -162,11 +152,11 @@ async def test_bind_async_engine() -> None: async def test_close_twice_sync(psycopg_url: str) -> None: """Test that closing a session releases connection resources, but remains usable.""" component = SQLAlchemyComponent(url=psycopg_url, prefer_async=False) - async with Context() as ctx: - await component.start(ctx) - session = ctx.require_resource(Session) + async with Context(): + await component.start() + session = get_resource_nowait(Session) assert isinstance(session.bind, Engine) - pool = session.bind.pool + pool = cast(QueuePool, session.bind.pool) assert isinstance(pool, QueuePool) session.execute(text("SELECT 1")) assert pool.checkedout() == 1 @@ -178,14 +168,14 @@ async def test_close_twice_sync(psycopg_url: str) -> None: assert pool.checkedout() == 0 -async def test_close_twice_async(psycopg_url: str) -> None: +async def test_close_twice_async(psycopg_url_async: str) -> None: """Test that closing a session releases connection resources, but remains usable.""" - component = SQLAlchemyComponent(url=psycopg_url) - async with Context() as ctx: - await component.start(ctx) - session = ctx.require_resource(AsyncSession) + component = SQLAlchemyComponent(url=psycopg_url_async) + async with Context(): + await component.start() + session = get_resource_nowait(AsyncSession) assert isinstance(session.bind, AsyncEngine) - pool = session.bind.pool + pool = cast(QueuePool, session.bind.pool) assert isinstance(pool, AsyncAdaptedQueuePool) await session.execute(text("SELECT 1")) assert pool.checkedout() == 1 @@ -217,14 +207,14 @@ async def test_finish_commit(raise_exception: bool, tmp_path: Path) -> None: component = SQLAlchemyComponent( url={"drivername": "sqlite", "database": str(db_path)}, ) - with ExitStack() as stack: - async with Context() as ctx: - await component.start(ctx) - session = ctx.require_resource(Session) - session.execute(text("INSERT INTO foo (id) VALUES(3)")) - if raise_exception: - stack.enter_context(pytest.raises(Exception, match="dummy")) - raise Exception("dummy") + async with AsyncExitStack() as stack: + await stack.enter_async_context(Context()) + await component.start() + session = get_resource_nowait(Session) + session.execute(text("INSERT INTO foo (id) VALUES(3)")) + if raise_exception: + stack.enter_context(pytest.raises(Exception, match="dummy")) + raise Exception("dummy") rows = connection.execute(text("SELECT * FROM foo")).fetchall() assert len(rows) == (0 if raise_exception else 1) @@ -233,16 +223,15 @@ async def test_finish_commit(raise_exception: bool, tmp_path: Path) -> None: async def test_memory_leak() -> None: """Test that creating a session in a context does not leak memory.""" component = SQLAlchemyComponent(url="sqlite:///:memory:") - async with Context() as ctx: - await component.start(ctx) - ctx.require_resource(Session) + async with Context(): + await component.start() + get_resource_nowait(Session) - del ctx gc.collect() # needed on PyPy assert next((x for x in gc.get_objects() if isinstance(x, Context)), None) is None -async def test_session_event_sync(psycopg_url: str) -> None: +async def test_session_event_sync(psycopg_url_async: str) -> None: """Test that creating a session in a context does not leak memory.""" listener_session: Session | None = None listener_thread: Thread | None = None @@ -253,28 +242,29 @@ def listener(session: Session) -> None: listener_session = session listener_thread = current_thread() - component = SQLAlchemyComponent(url=psycopg_url, prefer_async=False) - engine: Engine | None = None + component = SQLAlchemyComponent(url=psycopg_url_async, prefer_async=False) + engine: Engine try: - async with Context() as ctx: - await component.start(ctx) - engine = ctx.require_resource(Engine) + async with Context(): + await component.start() + engine = get_resource_nowait(Engine) Person.metadata.create_all(engine) - session_factory = ctx.require_resource(sessionmaker) + session_factory = get_resource_nowait(sessionmaker) listen(session_factory, "before_commit", listener) - dbsession = ctx.require_resource(Session) + dbsession = get_resource_nowait(Session) dbsession.add(Person(name="Test person")) assert listener_session is dbsession assert listener_thread != current_thread() finally: - if engine: - with engine.connect() as conn: - clear_database(conn) + with engine.connect() as conn: + clear_database(conn) -async def test_session_event_async(request: FixtureRequest, psycopg_url: str) -> None: +async def test_session_event_async( + request: FixtureRequest, psycopg_url_async: str +) -> None: """Test that creating a session in a context does not leak memory.""" listener_session: Session | None = None listener_thread: Thread | None = None @@ -282,7 +272,7 @@ async def test_session_event_async(request: FixtureRequest, psycopg_url: str) -> def listener(session: Session) -> None: nonlocal listener_session, listener_thread try: - async_session = get_resource(AsyncSession) + async_session = get_resource_nowait(AsyncSession) except NoCurrentContext: return @@ -292,13 +282,13 @@ def listener(session: Session) -> None: listen(Session, "before_commit", listener) request.addfinalizer(lambda: remove(Session, "before_commit", listener)) - component = SQLAlchemyComponent(url=psycopg_url) + component = SQLAlchemyComponent(url=psycopg_url_async) engine: AsyncEngine | None = None try: - async with Context() as ctx: - await component.start(ctx) - engine = ctx.require_resource(AsyncEngine) - dbsession = ctx.require_resource(AsyncSession) + async with Context(): + await component.start() + engine = get_resource_nowait(AsyncEngine) + dbsession = get_resource_nowait(AsyncSession) await dbsession.run_sync( lambda session: Person.metadata.create_all(session.bind) ) diff --git a/tests/test_testing_recipe.py b/tests/test_testing_recipe.py index 26fa962..8acf1e1 100644 --- a/tests/test_testing_recipe.py +++ b/tests/test_testing_recipe.py @@ -9,14 +9,14 @@ from typing import Any import pytest -from asphalt.core import ContainerComponent, Context +from asphalt.core import Component, Context, get_resource_nowait, start_component from sqlalchemy.engine import Connection, Engine from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession from sqlalchemy.orm import Session from sqlalchemy.sql.expression import delete, func, select -from asphalt.sqlalchemy.utils import clear_async_database, clear_database +from asphalt.sqlalchemy import clear_async_database, clear_database from .model import Base, Person @@ -49,8 +49,8 @@ def person(self, connection: Connection) -> Person: return person @pytest.fixture - def root_component(self, connection: Connection) -> ContainerComponent: - return ContainerComponent({"sqlalchemy": {"bind": connection}}) + def component_config(self, connection: Connection) -> dict[str, Any]: + return {"components": {"sqlalchemy": {"bind": connection}}} @pytest.fixture def dbsession(self, connection: Connection) -> Generator[Session, Any, None]: @@ -59,13 +59,13 @@ def dbsession(self, connection: Connection) -> Generator[Session, Any, None]: yield session async def test_rollback( - self, dbsession: Session, root_component: ContainerComponent + self, dbsession: Session, component_config: dict[str, Any] ) -> None: # Simulate a rollback happening in a subcontext - async with Context() as root_ctx: - await root_component.start(root_ctx) - async with Context() as ctx: - session = ctx.require_resource(Session) + async with Context(): + await start_component(Component, component_config) + async with Context(): + session = get_resource_nowait(Session) try: # No value for a non-nullable column => IntegrityError! session.add(Person()) @@ -77,30 +77,30 @@ async def test_rollback( session.add(Person(name="Works now!")) session.flush() - # The context is gone, but the extra Person should still be around - assert dbsession.scalar(func.count(Person.id)) == 2 + # The context is gone, but the extra Person should still be around + assert dbsession.scalar(func.count(Person.id)) == 2 async def test_add_person( - self, dbsession: Session, root_component: ContainerComponent + self, dbsession: Session, component_config: dict[str, Any] ) -> None: # Simulate adding a row to the "people" table in the application - async with Context() as root_ctx: - await root_component.start(root_ctx) - async with Context() as ctx: - session = ctx.require_resource(Session) + async with Context(): + await start_component(Component, component_config) + async with Context(): + session = get_resource_nowait(Session) session.add(Person(name="Another person")) # The testing code should see both rows now assert dbsession.scalar(func.count(Person.id)) == 2 async def test_delete_person( - self, dbsession: Session, root_component: ContainerComponent + self, dbsession: Session, component_config: dict[str, Any] ) -> None: # Simulate removing the test person in the application - async with Context() as root_ctx: - await root_component.start(root_ctx) - async with Context() as ctx: - session = ctx.require_resource(Session) + async with Context(): + await start_component(Component, component_config) + async with Context(): + session = get_resource_nowait(Session) session.execute(delete(Person)) # The testing code should not see any rows now @@ -135,8 +135,8 @@ async def person(self, connection: AsyncConnection) -> Person: return person @pytest.fixture - def root_component(self, connection: AsyncConnection) -> ContainerComponent: - return ContainerComponent({"sqlalchemy": {"bind": connection}}) + def component_config(self, connection: Connection) -> dict[str, Any]: + return {"components": {"sqlalchemy": {"bind": connection}}} @pytest.fixture async def dbsession( @@ -147,13 +147,13 @@ async def dbsession( yield session async def test_rollback( - self, dbsession: AsyncSession, root_component: ContainerComponent + self, dbsession: AsyncSession, component_config: dict[str, Any] ) -> None: # Simulate a rollback happening in a subcontext - async with Context() as root_ctx: - await root_component.start(root_ctx) - async with Context() as ctx: - session = ctx.require_resource(AsyncSession) + async with Context(): + await start_component(Component, component_config) + async with Context(): + session = get_resource_nowait(AsyncSession) try: # No value for a non-nullable column => IntegrityError! session.add(Person()) @@ -169,26 +169,26 @@ async def test_rollback( assert await dbsession.scalar(func.count(Person.id)) == 2 async def test_add_person( - self, dbsession: AsyncSession, root_component: ContainerComponent + self, dbsession: AsyncSession, component_config: dict[str, Any] ) -> None: # Simulate adding a row to the "people" table in the application - async with Context() as root_ctx: - await root_component.start(root_ctx) - async with Context() as ctx: - session = ctx.require_resource(AsyncSession) + async with Context(): + await start_component(Component, component_config) + async with Context(): + session = get_resource_nowait(AsyncSession) session.add(Person(name="Another person")) # The testing code should see both rows now assert await dbsession.scalar(select(func.count(Person.id))) == 2 async def test_delete_person( - self, dbsession: AsyncSession, root_component: ContainerComponent + self, dbsession: AsyncSession, component_config: dict[str, Any] ) -> None: # Simulate removing the test person in the application - async with Context() as root_ctx: - await root_component.start(root_ctx) - async with Context() as ctx: - session = ctx.require_resource(AsyncSession) + async with Context(): + await start_component(Component, component_config) + async with Context(): + session = get_resource_nowait(AsyncSession) await session.execute(delete(Person)) # The testing code should not see any rows now diff --git a/tests/test_utils.py b/tests/test_utils.py index 47b9b90..98edd05 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,15 +1,18 @@ from __future__ import annotations -from collections.abc import Generator +from collections.abc import AsyncGenerator, Generator from typing import Any import pytest from sqlalchemy.engine import Connection, Engine +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine from sqlalchemy.sql.ddl import CreateSchema, DropSchema from sqlalchemy.sql.schema import Column, ForeignKey, MetaData, Table from sqlalchemy.sql.sqltypes import Integer -from asphalt.sqlalchemy.utils import clear_database +from asphalt.sqlalchemy import clear_async_database, clear_database + +pytestmark = pytest.mark.anyio @pytest.fixture @@ -31,6 +34,27 @@ def connection(sync_engine: Engine) -> Generator[Connection, Any, None]: conn.execute(DropSchema("altschema")) +@pytest.fixture +async def async_connection( + async_engine: AsyncEngine, +) -> AsyncGenerator[AsyncConnection]: + async with async_engine.connect() as conn: + metadata = MetaData() + Table("table", metadata, Column("column1", Integer, primary_key=True)) + Table("table2", metadata, Column("fk_column", ForeignKey("table.column1"))) + if conn.dialect.name != "sqlite": + await conn.execute(CreateSchema("altschema")) + Table("table3", metadata, Column("fk_column", Integer), schema="altschema") + + await conn.run_sync(metadata.create_all) + + yield conn + + if conn.dialect.name != "sqlite": + await conn.run_sync(metadata.drop_all) + await conn.execute(DropSchema("altschema")) + + def test_clear_database(connection: Connection) -> None: clear_database( connection, ["altschema"] if connection.dialect.name != "sqlite" else [] @@ -43,3 +67,18 @@ def test_clear_database(connection: Connection) -> None: alt_metadata = MetaData(schema="altschema") alt_metadata.reflect(connection) assert len(alt_metadata.tables) == 0 + + +async def test_clear_async_database(async_connection: AsyncConnection) -> None: + await clear_async_database( + async_connection, + ["altschema"] if async_connection.dialect.name != "sqlite" else [], + ) + metadata = MetaData() + await async_connection.run_sync(metadata.reflect) + assert len(metadata.tables) == 0 + + if async_connection.dialect.name != "sqlite": + alt_metadata = MetaData(schema="altschema") + await async_connection.run_sync(alt_metadata.reflect) + assert len(alt_metadata.tables) == 0