Database & Persistence

Lexigram uses a protocol-first approach to data persistence. By abstracting the database layer behind the Repository Pattern, your domain logic stays clean and testable.

For complete API references and advanced performance tuning, see the lexigram-sql package.

---

1. The Repository Pattern

Instead of leaking ORM details into your services, you define Protocols (Contracts) that describe how data is accessed.

Defining the Protocol

from typing import Protocol, runtime_checkable
from lexigram.result import Result
from my_app.domain.models import Product

@runtime_checkable
class ProductRepository(Protocol):
    async def find(self, product_id: str) -> Result[Product, str]: ...
    async def save(self, product: Product) -> Result[None, str]: ...

Implementing with SQL

In your infrastructure layer, implement the protocol using your preferred database client.

from lexigram import singleton
from lexigram.result import Result
from my_app.domain.models import Product
from my_app.repositories.base import ProductRepository

@singleton
class SqlProductRepository(ProductRepository):
    """SQLAlchemy implementation of the ProductRepository contract."""

    async def find(self, product_id: str) -> Result[Product, str]:
        # Implementation here
        pass

    async def save(self, product: Product) -> Result[None, str]:
        # Implementation here
        pass

Registering with DI

Use a Service Provider to bind your implementation to the protocol. This allows you to swap the real database for an in-memory version during testing.

async def register(self, container: ContainerRegistrarProtocol) -> None:
    container.singleton(ProductRepository, SqlProductRepository)

For more on registration patterns, see Dependency Injection.

---

2. Model Definitions

Lexigram models are standard dataclasses or SQLAlchemy models that mirror your domain entities.

from dataclasses import dataclass

@dataclass
class Product:
    id: str
    name: str
    stock: int = 0

---

3. Database Configuration

Configure your database connection in application.yaml:

database:
  url: "postgresql+asyncpg://user:pass@localhost/dbname"
  echo: false
  pool_size: 20
  pool_timeout: 30

---

4. Async Connection Pattern

Lexigram uses async database connections throughout:

from lexigram import Application
from lexigram.config import LexigramConfig

async def main():
    config = LexigramConfig.from_yaml()
    db_config = config.get_section("database", DatabaseConfig)

    async with Application.boot(config=config) as app:
        db = await app.container.resolve(DatabaseProviderProtocol)
        # Use db.session() for scoped transactions

---

5. Transaction Management

Use scoped sessions for request-bound transactions:

async def register(self, container: ContainerRegistrarProtocol) -> None:
    container.scoped(DbSession, create_session_factory())

Then resolve within a scope:

async with container.scope() as scoped:
    session = await scoped.resolve(DbSession)
    result = await session.execute("SELECT * FROM products")