spakky-sqlalchemy

SQLAlchemy 통합 - ORM, Repository, Transaction, Outbox Contribution

spakky-sqlalchemy는 spakky-data의 Repository/Transaction 계약을 SQLAlchemy engine과 session으로 구현하는 플러그인입니다. 처음 사용하는 경우에는 데이터베이스 가이드에서 도메인 Aggregate, ORM table, Repository, @Transactional() UseCase를 한 흐름으로 확인하세요.

Outbox storage/table은 base plugin 본체가 아니라 spakky.contributions.spakky.outbox contribution으로 등록됩니다. load_plugins(include=...) 사용 시 spakky-outbox와 spakky-sqlalchemy가 모두 include set에 있어야 이 contribution이 로드됩니다. 설치 시에는 spakky-outbox를 별도로 추가하거나 spakky-sqlalchemy[outbox] extra를 사용하세요.

Agent state/signal/evidence persistence도 base plugin 본체가 아니라 spakky.contributions.spakky.agent contribution으로 등록됩니다. Durable @Agent execution을 운영에서 사용하려면 spakky-agent와 spakky-sqlalchemy[agent]를 함께 설치하고 두 plugin을 모두 active 상태로 로드해야 합니다. 이 contribution은 SqlAlchemyAgentStateRepository, SqlAlchemyAgentSignalRepository, SqlAlchemyAgentEvidenceRepository와 AgentStateTable, AgentSignalTable, AgentEvidenceTable을 등록합니다. 운영용 in-memory fallback은 제공하지 않습니다.

메인

spakky.plugins.sqlalchemy.main

initialize(app)

Initialize the SQLAlchemy plugin.

Registers SQLAlchemy configuration, schema registry, session managers, and transaction handlers. Async Pods are only registered when support_async_mode is True in the configuration.

Parameters:

Name	Type	Description	Default
app	SpakkyApplication	The Spakky application instance.	required

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize the SQLAlchemy plugin.

    Registers SQLAlchemy configuration, schema registry, session managers,
    and transaction handlers. Async Pods are only registered when
    support_async_mode is True in the configuration.

    Args:
        app: The Spakky application instance.
    """
    config = SQLAlchemyConnectionConfig()

    app.add(SQLAlchemyConnectionConfig)
    app.add(SchemaRegistry)

    app.add(ConnectionManager)
    app.add(SessionManager)
    app.add(Transaction)

    if config.support_async_mode:
        app.add(AsyncConnectionManager)
        app.add(AsyncSessionManager)
        app.add(AsyncTransaction)

options: show_root_heading: false

설정

spakky.plugins.sqlalchemy.common.config

SQLAlchemyConnectionConfig()

Bases: BaseSettings

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/common/config.py

def __init__(self) -> None:
    super().__init__()

connection_string instance-attribute

SQLAlchemy database connection URL (e.g. postgresql+psycopg://user:pass@host/db).

echo = False class-attribute instance-attribute

If True, the engine logs all SQL statements to the default logger.

echo_pool = False class-attribute instance-attribute

If True, the connection pool logs all checkout/checkin events.

DEFAULT_POOL_SIZE = 5 class-attribute

Default number of connections in the pool.

DEFAULT_POOL_MAX_OVERFLOW = 10 class-attribute

Default maximum overflow connections.

DEFAULT_POOL_TIMEOUT = 30.0 class-attribute

Default seconds to wait when pool is exhausted.

DEFAULT_POOL_RECYCLE = -1 class-attribute

Default connection recycle time (-1 = disabled).

DEFAULT_POOL_PRE_PING = False class-attribute

Default pre-ping setting.

pool_size = DEFAULT_POOL_SIZE class-attribute instance-attribute

Number of connections to maintain persistently in the pool.

pool_max_overflow = DEFAULT_POOL_MAX_OVERFLOW class-attribute instance-attribute

Maximum number of connections that can be opened beyond pool_size.

pool_timeout = DEFAULT_POOL_TIMEOUT class-attribute instance-attribute

Seconds to wait before raising an error when the pool is exhausted.

pool_recycle = DEFAULT_POOL_RECYCLE class-attribute instance-attribute

Recycle connections after this many seconds to prevent stale connections. Recommended when connecting through a proxy or firewall with idle timeouts.

pool_pre_ping = DEFAULT_POOL_PRE_PING class-attribute instance-attribute

If True, tests each connection for liveness before returning it from the pool. Prevents errors caused by connections dropped by the database server.

session_autoflush = True class-attribute instance-attribute

If True, the session automatically flushes pending changes before queries.

session_expire_on_commit = True class-attribute instance-attribute

If True, ORM objects are expired after each commit, forcing a reload on next access.

autocommit = True class-attribute instance-attribute

If True, transactions are automatically committed after with statements. If False, transactions must be manually committed or rolled back.

support_async_mode = True class-attribute instance-attribute

If True, registers async Pods (AsyncSessionManager, AsyncTransaction). Set to False when using database drivers that don't support async operations.

options: show_root_heading: false

ORM

spakky.plugins.sqlalchemy.orm.table

AbstractTable

Bases: DeclarativeBase, AsyncAttrs

Base table class for SQLAlchemy ORM integration with Spakky.

Use this class for infrastructure tables that don't map to domain models (e.g., outbox tables, audit logs). For tables that map to domain entities, use AbstractMappableTable instead.

AbstractMappableTable

Bases: AbstractTable

Table class with domain object mapping support.

Use this class for tables that map to domain entities. Subclasses must implement from_domain() and to_domain() methods for bidirectional mapping.

options: show_root_heading: false

spakky.plugins.sqlalchemy.orm.schema_registry

SchemaRegistry()

Bases: ITagRegistryAware

Registry that maps domain types to SQLAlchemy table schemas.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/orm/schema_registry.py

def __init__(self) -> None:
    self._metadata = MetaData()
    self._domain_to_table_map = {}
    self._table_to_domain_map = {}

metadata property

Return MetaData containing only tables registered via @Table tag.

This filters out tables that were defined but not scanned by the SpakkyApplication, ensuring only application-scoped tables are included.

Returns:

Type	Description
MetaData	MetaData with only @Table-tagged tables.

get_type(domain_type)

Look up the table class registered for the given domain type.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/orm/schema_registry.py

def get_type[AggregateRootT: AbstractAggregateRoot[Any]](
    self, domain_type: type[AggregateRootT]
) -> type[AbstractMappableTable[AggregateRootT]]:
    """Look up the table class registered for the given domain type."""
    table = self._domain_to_table_map.get(domain_type)
    if table is None:
        raise NoSchemaFoundFromDomainError(domain_type)
    return cast(type[AbstractMappableTable[AggregateRootT]], table)

from_domain(aggregate)

Convert a domain object to its corresponding table instance.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/orm/schema_registry.py

def from_domain[AggregateRootT: AbstractAggregateRoot[Any]](
    self, aggregate: AggregateRootT
) -> AbstractMappableTable[AggregateRootT]:
    """Convert a domain object to its corresponding table instance."""
    table: type[AbstractMappableTable[AggregateRootT]] | None = (
        self._domain_to_table_map.get(type(aggregate))
    )
    if table is None:
        raise NoSchemaFoundFromDomainError(aggregate)
    return table.from_domain(aggregate)

options: show_root_heading: false

spakky.plugins.sqlalchemy.orm.error

AbstractSpakkySqlAlchemyORMError

Bases: AbstractSpakkySqlAlchemyError, ABC

Base exception for Spakky SQLAlchemy ORM errors.

options: show_root_heading: false

영속성

spakky.plugins.sqlalchemy.persistency.repository

CannotDetermineAggregateTypeError

Bases: AbstractSpakkySqlAlchemyPersistencyError

Raised when aggregate type cannot be resolved from generic parameters.

AbstractGenericRepository(session_manager, schema_registry, aggregate_collector)

Bases: IGenericRepository[AggregateRootT, AggregateIdT], ABC

Generic repository implementation for SQLAlchemy.

This class provides basic CRUD operations using SQLAlchemy sessions. It serves as a base class for specific entity repositories.

Resolve aggregate type from generic parameters and store dependencies.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/repository.py

def __init__(
    self,
    session_manager: SessionManager,
    schema_registry: SchemaRegistry,
    aggregate_collector: AggregateCollector,
) -> None:
    """Resolve aggregate type from generic parameters and store dependencies."""
    parameterized_base = next(
        (
            type_
            for type_ in generic_mro(type(self))
            if get_origin(type_) is AbstractGenericRepository
        ),
        None,
    )
    if parameterized_base is None:
        raise CannotDetermineAggregateTypeError(type(self))
    self._aggregate_type = next(iter(get_args(parameterized_base)))
    self._session_manager = session_manager
    self._schema_registry = schema_registry
    self._aggregate_collector = aggregate_collector

AbstractAsyncGenericRepository(session_manager, schema_registry, aggregate_collector)

Bases: IAsyncGenericRepository[AggregateRootT, AggregateIdT], ABC

Async generic repository implementation for SQLAlchemy.

This class provides basic CRUD operations using SQLAlchemy's async sessions. It serves as a base class for specific entity repositories that require async support.

Resolve aggregate type from generic parameters and store dependencies.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/repository.py

def __init__(
    self,
    session_manager: AsyncSessionManager,
    schema_registry: SchemaRegistry,
    aggregate_collector: AggregateCollector,
) -> None:
    """Resolve aggregate type from generic parameters and store dependencies."""
    parameterized_base = next(
        (
            type_
            for type_ in generic_mro(type(self))
            if get_origin(type_) is AbstractAsyncGenericRepository
        ),
        None,
    )
    if parameterized_base is None:
        raise CannotDetermineAggregateTypeError(type(self))
    self._aggregate_type = next(iter(get_args(parameterized_base)))
    self._session_manager = session_manager
    self._schema_registry = schema_registry
    self._aggregate_collector = aggregate_collector

options: show_root_heading: false

spakky.plugins.sqlalchemy.persistency.transaction

Transaction(config, session_manager)

Bases: AbstractTransaction

SQLAlchemy synchronous transaction implementation.

Initialize with autocommit config and session manager.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

def __init__(
    self,
    config: SQLAlchemyConnectionConfig,
    session_manager: SessionManager,
) -> None:
    """Initialize with autocommit config and session manager."""
    super().__init__(autocommit=config.autocommit)
    self._session_manager = session_manager

initialize()

Open a new session for this transaction scope.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
def initialize(self) -> None:
    """Open a new session for this transaction scope."""
    self._session_manager.open()

dispose()

Close the session after transaction completes.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
def dispose(self) -> None:
    """Close the session after transaction completes."""
    self._session_manager.close()

commit()

Commit the current session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
def commit(self) -> None:
    """Commit the current session."""
    self._session_manager.session.commit()

rollback()

Roll back the current session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
def rollback(self) -> None:
    """Roll back the current session."""
    self._session_manager.session.rollback()

AsyncTransaction(config, session_manager)

Bases: AbstractAsyncTransaction

SQLAlchemy asynchronous transaction implementation.

Initialize with autocommit config and async session manager.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

def __init__(
    self,
    config: SQLAlchemyConnectionConfig,
    session_manager: AsyncSessionManager,
) -> None:
    """Initialize with autocommit config and async session manager."""
    super().__init__(autocommit=config.autocommit)
    self._session_manager = session_manager

initialize() async

Open a new async session for this transaction scope.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
async def initialize(self) -> None:
    """Open a new async session for this transaction scope."""
    await self._session_manager.open()

dispose() async

Close the async session after transaction completes.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
async def dispose(self) -> None:
    """Close the async session after transaction completes."""
    await self._session_manager.close()

commit() async

Commit the current async session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
async def commit(self) -> None:
    """Commit the current async session."""
    await self._session_manager.session.commit()

rollback() async

Roll back the current async session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/transaction.py

@override
async def rollback(self) -> None:
    """Roll back the current async session."""
    await self._session_manager.session.rollback()

options: show_root_heading: false

spakky.plugins.sqlalchemy.persistency.session_manager

SessionNotInitializedError

Bases: AbstractSpakkySqlAlchemyPersistencyError

Raised when trying to access a session that has not been initialized.

SessionManager(connection_manager)

Bases: IApplicationContextAware

Manages scoped synchronous SQLAlchemy sessions.

Initialize with engine from connection manager.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/session_manager.py

def __init__(self, connection_manager: ConnectionManager) -> None:
    """Initialize with engine from connection manager."""
    self._engine = connection_manager.connection
    self._current_session = None

open()

Open a new scoped session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/session_manager.py

def open(self) -> None:
    """Open a new scoped session."""
    self._current_session = self._scoped_session()

close()

Close the current session and remove the scoped session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/session_manager.py

def close(self) -> None:
    """Close the current session and remove the scoped session."""
    if self._current_session is not None:
        self._current_session.close()
    self._scoped_session.remove()
    self._current_session = None

AsyncSessionManager(connection_manager)

Bases: IApplicationContextAware

Manages scoped asynchronous SQLAlchemy sessions.

Initialize with engine from async connection manager.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/session_manager.py

def __init__(self, connection_manager: AsyncConnectionManager) -> None:
    """Initialize with engine from async connection manager."""
    self._engine = connection_manager.connection
    self._current_session = None

open() async

Open a new scoped async session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/session_manager.py

async def open(self) -> None:
    """Open a new scoped async session."""
    self._current_session = self._scoped_session()

close() async

Close the current session and remove the scoped async session.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/session_manager.py

async def close(self) -> None:
    """Close the current session and remove the scoped async session."""
    if self._current_session is not None:
        await self._current_session.close()
    await self._scoped_session.remove()
    self._current_session = None

options: show_root_heading: false

spakky.plugins.sqlalchemy.persistency.connection_manager

Connection managers for SQLAlchemy engines.

This module provides connection managers that handle SQLAlchemy engine lifecycle, including proper disposal on application shutdown.

ConnectionManager(config)

Manages synchronous SQLAlchemy engine lifecycle.

Create a synchronous SQLAlchemy engine from configuration.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/connection_manager.py

def __init__(self, config: SQLAlchemyConnectionConfig) -> None:
    """Create a synchronous SQLAlchemy engine from configuration."""
    self._connection = create_engine(
        url=config.connection_string,
        echo=config.echo,
        echo_pool=config.echo_pool,
        pool_size=config.pool_size,
        max_overflow=config.pool_max_overflow,
        pool_timeout=config.pool_timeout,
        pool_recycle=config.pool_recycle,
        pool_pre_ping=config.pool_pre_ping,
    )

connection property

Get the SQLAlchemy Engine instance.

dispose()

Dispose the engine connection pool.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/connection_manager.py

def dispose(self) -> None:
    """Dispose the engine connection pool."""
    self._connection.dispose()

AsyncConnectionManager(config)

Manages asynchronous SQLAlchemy engine lifecycle.

Note: Does not implement IAsyncService because AsyncEngine.dispose() must be called from the same event loop where connections were created. In test environments with pytest-asyncio, this means dispose must be called from the test's event loop, not ApplicationContext's internal event loop thread.

Create an asynchronous SQLAlchemy engine from configuration.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/connection_manager.py

def __init__(self, config: SQLAlchemyConnectionConfig) -> None:
    """Create an asynchronous SQLAlchemy engine from configuration."""
    self._connection = create_async_engine(
        url=config.connection_string,
        echo=config.echo,
        echo_pool=config.echo_pool,
        pool_size=config.pool_size,
        max_overflow=config.pool_max_overflow,
        pool_timeout=config.pool_timeout,
        pool_recycle=config.pool_recycle,
        pool_pre_ping=config.pool_pre_ping,
    )

connection property

Get the async SQLAlchemy Engine instance.

dispose() async

Dispose the engine connection pool.

Must be called from the same event loop where connections were used.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/persistency/connection_manager.py

async def dispose(self) -> None:
    """Dispose the engine connection pool.

    Must be called from the same event loop where connections were used.
    """
    await self._connection.dispose()

options: show_root_heading: false

spakky.plugins.sqlalchemy.persistency.error

AbstractSpakkySqlAlchemyPersistencyError

Bases: AbstractSpakkySqlAlchemyError, ABC

Base exception for Spakky SQLAlchemy persistency errors.

options: show_root_heading: false

Outbox

spakky.plugins.sqlalchemy.contributions.outbox

SQLAlchemy contribution for the spakky-outbox feature.

initialize(app)

Register SQLAlchemy-backed Outbox infrastructure.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/contributions/outbox.py

def initialize(app: SpakkyApplication) -> None:
    """Register SQLAlchemy-backed Outbox infrastructure."""
    config = SQLAlchemyConnectionConfig()

    app.add(OutboxMessageTable)
    app.add(SqlAlchemyOutboxStorage)

    if config.support_async_mode:
        app.add(AsyncSqlAlchemyOutboxStorage)

options: show_root_heading: false

spakky.plugins.sqlalchemy.outbox.storage

SQLAlchemy implementation of IOutboxStorage / IAsyncOutboxStorage.

SqlAlchemyOutboxStorage(session_manager, connection_manager, config=None)

Bases: IOutboxStorage

Synchronous SQLAlchemy-based Outbox storage implementation.

• save(): uses the current transactional session (same TX as business data).
• fetch_pending/mark_published/increment_retry: use independent sessions.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/outbox/storage.py

def __init__(
    self,
    session_manager: SessionManager,
    connection_manager: ConnectionManager,
    config: OutboxConfig | None = None,
) -> None:
    self._session_manager = session_manager
    self._session_factory = sessionmaker(
        bind=connection_manager.connection,
        expire_on_commit=False,
    )
    self._claim_timeout_seconds = (
        config.claim_timeout_seconds if config else _DEFAULT_CLAIM_TIMEOUT_SECONDS
    )

AsyncSqlAlchemyOutboxStorage(session_manager, connection_manager, config=None)

Bases: IAsyncOutboxStorage

Asynchronous SQLAlchemy-based Outbox storage implementation.

• save(): uses the current transactional session (same TX as business data).
• fetch_pending/mark_published/increment_retry: use independent sessions.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/outbox/storage.py

def __init__(
    self,
    session_manager: AsyncSessionManager,
    connection_manager: AsyncConnectionManager,
    config: OutboxConfig | None = None,
) -> None:
    self._session_manager = session_manager
    self._session_factory = async_sessionmaker(
        bind=connection_manager.connection,
        expire_on_commit=False,
    )
    self._claim_timeout_seconds = (
        config.claim_timeout_seconds if config else _DEFAULT_CLAIM_TIMEOUT_SECONDS
    )

options: show_root_heading: false

spakky.plugins.sqlalchemy.outbox.table

SQLAlchemy table definition for the Outbox pattern.

OutboxMessageTable

Bases: AbstractTable

Outbox message table for transactional outbox pattern.

This is an infrastructure table that doesn't map to a domain model, so it inherits from AbstractTable (not AbstractMappableTable).

options: show_root_heading: false

Agent Persistence

spakky.plugins.sqlalchemy.contributions.agent

SQLAlchemy contribution for the spakky-agent feature.

initialize(app)

Register SQLAlchemy-backed agent persistence infrastructure.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/contributions/agent.py

def initialize(app: SpakkyApplication) -> None:
    """Register SQLAlchemy-backed agent persistence infrastructure."""
    app.add(AgentStateTable)
    app.add(AgentSignalTable)
    app.add(AgentEvidenceTable)
    app.add(SqlAlchemyAgentStateRepository)
    app.add(SqlAlchemyAgentSignalRepository)
    app.add(SqlAlchemyAgentEvidenceRepository)

options: show_root_heading: false

spakky.plugins.sqlalchemy.agent.repository

SQLAlchemy repository implementations for spakky-agent contracts.

SqlAlchemyAgentStateRepository(session_manager)

Bases: IAgentStateRepository

SQLAlchemy materialized state repository for agent executions.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/agent/repository.py

def __init__(self, session_manager: SessionManager) -> None:
    self._session_manager = session_manager

SqlAlchemyAgentSignalRepository(session_manager)

Bases: IAgentSignalRepository

SQLAlchemy durable inbound queue repository for agent signals.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/agent/repository.py

def __init__(self, session_manager: SessionManager) -> None:
    self._session_manager = session_manager

SqlAlchemyAgentEvidenceRepository(session_manager)

Bases: IAgentEvidenceRepository

SQLAlchemy append-only repository for agent evidence artifacts.

Source code in plugins/spakky-sqlalchemy/src/spakky/plugins/sqlalchemy/agent/repository.py

def __init__(self, session_manager: SessionManager) -> None:
    self._session_manager = session_manager

options: show_root_heading: false

spakky.plugins.sqlalchemy.agent.table

SQLAlchemy schema mappings for spakky-agent persistence contracts.

AgentStateTable

Bases: AbstractMappableTable[AgentState]

Materialized state table for durable agent executions.

AgentSignalTable

Bases: AbstractMappableTable[AgentSignal]

Durable inbound signal queue table for agent executions.

AgentEvidenceTable

Bases: AbstractMappableTable[AgentEvidence]

Append-only evidence table for agent execution artifacts.

options: show_root_heading: false

에러

spakky.plugins.sqlalchemy.error

AbstractSpakkySqlAlchemyError

Bases: AbstractSpakkyFrameworkError, ABC

Base exception for Spakky SQLAlchemy errors.

options: show_root_heading: false

추가 모듈

Errors for SQLAlchemy-backed agent persistence.

AgentPersistenceRowNotFoundError

Bases: AbstractSpakkySqlAlchemyPersistencyError

Raised when an agent persistence row cannot be found.