# Database Adapters

`fansifter_common.adapters.db` provides SQLAlchemy-based database access with session management, transactions, ORM models, and a chainable query builder.

## Setup

```python
from fansifter_common.adapters.db import Database

db = Database(url="postgresql://user:pass@localhost/dbname")

# With extra SQLAlchemy engine/session options
db = Database(
    url="postgresql://user:pass@localhost/dbname",
    engine_args={"echo": True, "pool_size": 10},
    session_args={"autoflush": True},
)
```

Close the connection pool when shutting down:

```python
db.close()
```

Check connectivity:

```python
from fansifter_common.adapters.db.utils import check_db_alive

if not check_db_alive(db.engine.url):
    raise RuntimeError("Database unreachable")
```

---

## Sessions

A session must be active before any database work. Use `session_factory` as a context manager:

```python
with db.session_factory():
    # db.session is available here
    result = db.session.execute(sa.text("SELECT 1"))
```

If a session is already open, `session_factory` reuses it by default. Pass `replace=True` to force a new one:

```python
with db.session_factory(replace=True):
    ...
```

---

## Transactions

`db.transaction()` manages its own session — no outer `session_factory` needed.

### Context manager

```python
with db.transaction():
    artist = Artist(name="Alice")
    artist.save()
# committed here
```

### Decorator

```python
@db.transaction
def create_artist(name: str) -> Artist:
    artist = Artist(name=name)
    artist.save()
    return artist
```

### Commit on specific errors

Useful when you want to persist state even after a known exception (e.g. recording a failed job):

```python
with db.transaction(commit_on_error=ValueError):
    do_something_that_might_raise()
```

### AUTOCOMMIT mode

Use `autocommit` for **read-only** operations or DDL statements that must run outside a transaction (e.g. `VACUUM`, `CREATE INDEX CONCURRENTLY`). Do not use it for writes — there is no rollback if something goes wrong.

To avoid a spurious rollback being issued when the session closes, set `skip_autocommit_rollback` in your engine args:

```python
db = Database(
    url="postgresql://user:pass@localhost/dbname",
    engine_args={"skip_autocommit_rollback": True},
)
```

```python
with db.autocommit():
    result = db.session.execute(sa.text("SELECT 1"))
```

Or as a decorator:

```python
@db.autocommit
def get_server_version() -> str:
    return db.session.execute(sa.text("SELECT version()")).scalar()
```

---

## Models

### Using `Model` (single-project setup)

If you only need one database and don't need a custom `type_annotation_map`, inherit from `Model` directly:

```python
from sqlalchemy.orm import Mapped, mapped_column

from fansifter_common.adapters.db.models import Model


class Artist(Model):
    __tablename__ = "artists"
    __db__ = db

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str]
```

### Using `ModelMixin` (multi-project / custom type map)

When you need your own registry, `type_annotation_map`, or per-project `__db__`, use `ModelMixin` to compose your own base:

```python
import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase, MappedAsDataclass

from fansifter_common.adapters.db.models import ModelMixin
from fansifter_common.adapters.db.types import NaiveUTCDateTime

# db must be instantiated before the Model class is defined
db = Database(url="postgresql://user:pass@localhost/dbname")


class Model(ModelMixin, MappedAsDataclass, DeclarativeBase):
    __db__ = db
    type_annotation_map = {
        str: sa.Text,
        datetime: NaiveUTCDateTime,
    }
```

All models in the project then inherit from this local `Model` and get `save`/`delete`/`refresh`/`query` for free.

### Autodiscovery

SQLAlchemy only knows about models that have been imported. Call `autodiscover_models` at startup to ensure all model modules are loaded before `create_all` or Alembic runs:

```python
from fansifter_common.adapters.db.models import autodiscover_models

autodiscover_models("myproject")
```

This recursively imports every submodule of `myproject` whose name is `models` or contains `.models.`. The `search` parameter lets you target a different naming convention:

```python
autodiscover_models("myproject", search="entities")
```

### Save, delete, refresh

```python
with db.transaction():
    artist = Artist(name="Alice", email="alice@example.com")
    artist.save()  # flushes inside a transaction, commits otherwise
    artist.refresh()
    artist.delete()
```

Pass `flush=False` to defer flushing within a transaction:

```python
with db.transaction():
    artist.save(flush=False)
    # ... batch more work ...
    # committed at end of `with` block
```

---

## Querying

`Model.query` is a chainable query builder tied to the active session.

### Fetch all / first / one

```python
with db.session_factory():
    artists = Artist.query.all()
    artist = Artist.query.first()
    artist = Artist.query.one()  # raises ObjectNotFoundError if 0 or >1 results
    artist = Artist.query.one_or_none()  # None if not found
```

### Filter

```python
with db.session_factory():
    active_artists = Artist.query.where(Artist.is_active == True).all()
    artist = Artist.query.where(Artist.email == "alice@example.com").one_or_none()
```

### Get by primary key

```python
with db.session_factory():
    artist = Artist.query.get(42)  # None if not found
    artist = Artist.query.get_one(42)  # raises ObjectNotFoundError if not found
```

### Ordering, limit, offset

```python
with db.session_factory():
    page = Artist.query.order_by(Artist.created_at.desc()).limit(20).offset(40).all()
```

### Most recent row

```python
with db.session_factory():
    latest = Artist.query.latest(Artist.created_at)
```

### Count and exists

```python
with db.session_factory():
    total = Artist.query.where(Artist.is_active == True).count()
    exists = Artist.query.where(Artist.email == "alice@example.com").exists()
```

### Joins

```python
with db.session_factory():
    artists = (
        Artist.query.join(Order, Order.artist_id == Artist.id)
        .where(Order.status == "pending")
        .all()
    )

    # Left outer join
    artists = Artist.query.join(Order, Order.artist_id == Artist.id, isouter=True).all()
```

### Eager loading relationships

```python
with db.session_factory():
    artists = Artist.query.joinedload(Artist.albums).all()

    # Nested eager load
    artists = Artist.query.joinedload(Artist.albums, Album.tracks).all()
```

### Row-level locking

```python
with db.transaction():
    artist = Artist.query.where(Artist.id == 42).get(42, with_for_update=True)
```

### Custom query class

Subclass `Query` to add domain-specific query methods, then attach it to the model via `as_descriptor()`:

```python
from typing import Self

from fansifter_common.adapters.db.models import Query


class ArtistQuery(Query["Artist"]):
    def active(self) -> Self:
        return self.where(Artist.is_active == True)

    def by_genre(self, genre: str) -> Self:
        return self.where(Artist.genre == genre)


class Artist(Model):
    __tablename__ = "artists"
    __db__ = db

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    genre: Mapped[str]
    is_active: Mapped[bool]

    query = ArtistQuery.as_descriptor()
```

Custom methods return `Self`, so they chain naturally with the built-in builder methods:

```python
with db.session_factory():
    artists = Artist.query.active().by_genre("jazz").order_by(Artist.name).all()
```

---

## Custom SQL queries

For complex queries that don't fit the builder, use `db.session` directly with SQLAlchemy core or raw SQL.

### SQLAlchemy core

```python
import sqlalchemy as sa

with db.session_factory():
    stmt = sa.select(Artist).where(Artist.is_active == True).order_by(Artist.name)
    artists = db.session.execute(stmt).scalars().all()
```

### Raw SQL

```python
with db.session_factory():
    result = db.session.execute(
        sa.text("SELECT id, name FROM artists WHERE email = :email"),
        {"email": "alice@example.com"},
    )
    row = result.mappings().one_or_none()
```

### Jinja2 SQL templates

For dynamic queries (e.g. variable filters, optional clauses), use `query_from_template`. Templates are rendered via `Jinja2SQL` and the result is a `TextClause` ready for execution.

Pass a `Jinja2SQL` instance at construction time:

```python
from jinja2sql import Jinja2SQL

db = Database(url="...", jinja2sql=Jinja2SQL(path="path/to/sql/templates"))
```

Then render and execute:

```python
with db.session_factory():
    query = db.query_from_template(
        "users/search.sql.j2",
        context={"search": "%alice%", "order_by": "name.DESC"},
    )
    users = db.session.execute(query).scalars().all()
```

The `db_dialect` variable is automatically injected into every template context.

#### Built-in template filters

| Filter | Usage | Output |
|--------|-------|--------|
| `orderby` | `{{ order_by \| orderby }}` | `name DESC NULLS LAST` |
| `enum_values` | `{{ MyEnum \| enum_values }}` | `(0, 'active'), (1, 'inactive')` |
| `enum_choices` | `{{ MyEnum \| enum_choices }}` | same, for choice-based enums |
| `array_values` | `{{ MyEnum \| array_values }}` | `'active', 'inactive'` |

`orderby` accepts a string or list of strings in `"column[.direction][.nulls]"` format:

```sql
-- template
ORDER BY {{ order_by | orderby }}

-- order_by = "created_at.DESC.NULLS_LAST"  →  created_at DESC NULLS LAST
-- order_by = ["name", "created_at.DESC"]   →  name ASC, created_at DESC
```

---

## Custom column types

Import from `fansifter_common.adapters.db.types`:

| Type | Description |
|------|-------------|
| `SafeJSONType` | Stores arbitrary data as JSON text |
| `NaiveUTCDateTime` | Enforces timezone-aware input; stores as naive UTC |
| `PydanticType(MyModel)` | Serializes/validates a Pydantic model |
| `ChoiceType` | SQLAlchemy-utils `ChoiceType` stored as text |
| `EncryptedStrType` | Returns an `EncryptedStr` wrapper on read |

Example:

```python
from fansifter_common.adapters.db.types import (
    NaiveUTCDateTime,
    PydanticType,
    SafeJSONType,
)


class Event(Model):
    __tablename__ = "events"
    __db__ = db

    id: Mapped[int] = mapped_column(primary_key=True)
    occurred_at: Mapped[datetime] = mapped_column(NaiveUTCDateTime)
    payload: Mapped[dict] = mapped_column(SafeJSONType)
    metadata: Mapped[MySchema] = mapped_column(PydanticType(MySchema))
```

---

## Exceptions

`fansifter_common.adapters.db.exceptions.ObjectNotFoundError` (HTTP 404) is raised by:

- `Query.one()` — when no rows or more than one row match
- `Query.get_one(ident)` — when the primary key is not found

```python
from fansifter_common.adapters.db.exceptions import ObjectNotFoundError

try:
    artist = Artist.query.where(Artist.email == "x@example.com").one()
except ObjectNotFoundError:
    ...
```

---

## Testing

### Rollback-only transactions

`db.rollback_transaction()` opens a session and a connection that always rolls back at the end, regardless of any `save()` or `commit()` calls inside. Use it as a test fixture to keep the database clean between tests:

```python
with db.rollback_transaction():
    artist = Artist(name="Test")
    artist.save()
# rolled back — nothing persisted
```

> **Note:** `rollback_transaction` raises `RuntimeError` if a session is already open. Call it before any other session/transaction context.

### Sharing a session across threads

By default, `db.session` is stored in a `ContextVar`, so each thread gets its own isolated session. In tests that spawn threads but want them all to share the same session and see the same uncommitted rows, wrap the test body with `db.global_context()`:

```python
with db.global_context():
    with db.rollback_transaction():
        # all threads spawned here will share this session
        t = threading.Thread(target=some_db_work)
        t.start()
        t.join()
```

Without `global_context`, the spawned thread would have its own `ContextVar` slot with no session and raise `RuntimeError: Session is not started`.