Skip to content

Server Storage

Module: s2auth.server.storage

StoredObject

Bases: Base

Database model for storing pydantic objects as JSON.

Source code in src/s2auth/server/models.py 
 9
10
11
12
13
14
15
16
17
18
19
class StoredObject(Base):
    """Database model for storing pydantic objects as JSON."""

    __tablename__ = "stored_objects"

    id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
    timestamp_received = Column(
        DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)
    )
    object_type = Column(String, nullable=False, index=True)
    data = Column(Text, nullable=False)  # JSON stored as TEXT for SQLite compatibility

async_session(cfg=Depends[config]) async

Source code in src/s2auth/server/db.py 
 8
 9
10
11
12
13
14
15
16
17
18
@register_provider(context_manager=True)
@asynccontextmanager
async def async_session(cfg: Config = Depends[config]):
    engine = create_async_engine(cfg.sqlalchemy_db_uri.get_secret_value())
    session = AsyncSession(engine)
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise

store_object(object, async_session=Depends[async_session]) async

Store a pydantic object in the database as JSON.

Parameters:

Name Type Description Default
object BaseModel

Any pydantic BaseModel instance to store

 required
async_session AsyncSession

Database session (injected via dependency injection)

 Depends[async_session]
Source code in src/s2auth/server/storage.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@inject
async def store_object(
    object: BaseModel, async_session: AsyncSession = Depends[async_session]
) -> None:
    """Store a pydantic object in the database as JSON.

    Args:
        object: Any pydantic BaseModel instance to store
        async_session: Database session (injected via dependency injection)
    """
    # Get the object type name
    object_type = type(object).__name__

    # Convert pydantic object to JSON string
    json_data = object.model_dump_json()

    # Create database record (ID will be auto-generated)
    stored_obj = StoredObject(
        object_type=object_type,
        data=json_data,
    )

    # Add to session (will be committed by the async_session provider)
    async_session.add(stored_obj)