Metadata-Version: 2.4
Name: osm-db-session
Version: 0.0.3
Summary: Async SQLAlchemy session manager
Requires-Python: >=3.13
Description-Content-Type: text/markdown
Requires-Dist: alembic>=1.18.4
Requires-Dist: greenlet>=3.2.4
Requires-Dist: sqlalchemy>=2.0
Provides-Extra: testkit
Requires-Dist: pytest>=9.0.2; extra == "testkit"
Requires-Dist: pytest-asyncio>=1.3.0; extra == "testkit"

# osm-db-session

Менеджер асинхронных SQLAlchemy-сессий для Python 3.13+.

## Получение или создание записи

`DatabaseSessionManager.get_or_create()` ищет запись по именованным аргументам
и возвращает пару `(instance, created)`. Если запись найдена, `created` равен
`False`, её поля не меняются и `commit()` не вызывается. Если запись отсутствует,
метод создаёт её, выполняет `commit()` и возвращает `created=True`.

Пример для модели `Widget` вашего приложения с уникальным полем `name` и полем
`value`:

```python
from osm_db_session import DatabaseSessionManager
from myapp.models import Widget


async def ensure_widget(dsm: DatabaseSessionManager) -> None:
    widget, created = await dsm.get_or_create(
        Widget,
        name="example",
        defaults={"value": "initial"},
    )

    existing, created_again = await dsm.get_or_create(
        Widget,
        name="example",
        defaults={"value": "ignored"},
    )
    assert existing.id == widget.id
    assert existing.value == "initial"
    assert created_again is False
```

Поиск использует только именованные аргументы после `defaults`. При создании
они объединяются с `defaults`; значения из `defaults` имеют приоритет при
совпадении ключей. `defaults` можно не передавать, передать `None` или пустой
словарь. В отличие от `get_or_create()`, метод `update_or_create()` применяет
`defaults` и к уже существующей записи.

`get_or_create()` не гарантирует атомарность при конкурентном создании. Если
вставка нарушает ограничение уникальности, метод откатывает сессию и передаёт
`sqlalchemy.exc.IntegrityError` вызывающему коду без повторного поиска или
автоматического повтора операции.

## Инструменты для pytest

Опциональный набор инструментов создаёт уникальную базу данных PostgreSQL на
время запуска тестов и предоставляет движки, а также транзакционно изолированные
экземпляры `DatabaseSessionManager`. Приложение самостоятельно устанавливает
асинхронный драйвер PostgreSQL и запускает собственные миграции.

Установите пакет с дополнительными зависимостями для тестирования и драйвером,
например `asyncpg`:

```shell
uv add --dev "osm-db-session[testkit]" asyncpg
```

Extra `testkit` включает `pytest` и `pytest-asyncio`; драйвер БД в него не входит.

Подключите плагин в корневом `conftest.py` приложения:

```python
pytest_plugins = ["osm_db_session.testing.pytest_plugin"]
```

Плагин читает `DB_DSN`, а если переменная не задана или пуста — `DB__DSN`.
Значение должно быть асинхронным PostgreSQL URL существующей базы данных,
например `postgresql+asyncpg://user:password@localhost/app`. Пользователю БД
требуется право создания баз данных. База из DSN не удаляется: для запуска
создаётся отдельная база, имя которой содержит `_test_`, случайный токен и
идентификатор рабочего процесса pytest-xdist. После тестов эта база удаляется.
Загрузка плагина без обращения к его DB-фикстурам не требует DSN или работающей БД.

### Миграции

Проекты, использующие `migrated_db`, `db_engine` или `test_dsm`, должны
переопределить фикстуру `db_migrator` со scope `session`. Мигратор получает URL
созданной тестовой базы и вызывается один раз. Например, интеграцию с Alembic
можно настроить в корневом `conftest.py`:

```python
import os
from unittest.mock import patch

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy.engine import URL

from osm_db_session.testing import DBMigrator

pytest_plugins = ["osm_db_session.testing.pytest_plugin"]


@pytest.fixture(scope="session")
def db_migrator(pytestconfig: pytest.Config) -> DBMigrator:
    def migrate(dsn: URL) -> None:
        database_url = dsn.render_as_string(hide_password=False)
        config = Config(str(pytestconfig.rootpath / "alembic.ini"))
        config.set_main_option(
            "script_location",
            str(pytestconfig.rootpath / "migrations"),
        )
        config.set_main_option(
            "sqlalchemy.url",
            database_url.replace("%", "%%"),
        )

        # Нужно, если env.py проекта читает DSN из окружения напрямую.
        with patch.dict(
            os.environ,
            {"DB_DSN": database_url, "DB__DSN": database_url},
        ):
            command.upgrade(config, "head")

    return migrate
```

Пример ожидает `alembic.ini` и каталог `migrations` в корне проекта-потребителя.
Если `env.py` использует только URL из конфигурации Alembic, вызов
`command.upgrade(config, "head")` можно выполнять без блока `patch.dict`.

### Фикстуры

| Фикстура | Scope | Назначение |
| --- | --- | --- |
| `admin_db_dsn` | session | Разбирает исходный DSN из окружения. |
| `template_db_dsn` | session | Формирует URL уникальной тестовой базы. |
| `template_db` | session | Создаёт и удаляет пустую тестовую базу. |
| `db_migrator` | session | Задаёт функцию миграции, которую предоставляет приложение. |
| `migrated_db` | session | Вызывает мигратор и возвращает URL подготовленной базы. |
| `db_engine` | function | Предоставляет `AsyncEngine`; автоматически не откатывает изменения. |
| `test_dsm` | function | Предоставляет `TestDSM` внутри внешней транзакции, откатываемой после теста. |

`TestDSM` наследует методы `DatabaseSessionManager`, включая `get_or_create()`.
Откат внешней транзакции отменяет и изменения, для которых код приложения
вызвал `commit()`:

```python
from osm_db_session.testing import TestDSM
from myapp.models import Widget


async def test_create_widget(test_dsm: TestDSM) -> None:
    widget, created = await test_dsm.get_or_create(
        Widget,
        name="example",
        defaults={"value": "initial"},
    )
    assert created is True
    assert widget.value == "initial"
```

При самостоятельном создании `TestDSM` используйте его сессии внутри
`async with manager.rollback_tx()` и закрывайте менеджер через `await manager.close()`.

Для асинхронных тестов и фикстур приложения настройте pytest:

```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
```

При использовании strict-режима pytest-asyncio вместо `asyncio_mode = "auto"`
помечайте асинхронные тесты `@pytest.mark.asyncio`, а асинхронные фикстуры —
`@pytest_asyncio.fixture`.

## Проверка библиотеки

```shell
uv run --group test pytest
uv lock --check
uv build
```

Основные тесты используют SQLite и не требуют PostgreSQL. Интеграционный тест
жизненного цикла PostgreSQL запускается при заданном `DB_DSN`; для него отдельно
установите соответствующий драйвер и настройте тестовый сервер. Для запуска
только тестов без внешних сервисов используйте `pytest -m "not integration"`
через ту же команду `uv run --group test`.
