Skip to main content

Storage Adapters API

Concrete storage implementations that fulfill the protocol contracts defined in the ports layer.

chaoscypher_core.adapters.sqlite.adapter​

SQLite Storage Adapter for ChaosCypher Knowledge Engine.

Provides complete SQLite database storage implementation using SQLModel. Implements the per-domain storage Protocols defined under chaoscypher_core.ports (storage_workflows, storage_chats, storage_sources, ...).

This adapter requires SQLModel: pip install sqlmodel>=0.0.14

Usage: from chaoscypher_core.adapters.sqlite import SqliteAdapter

adapter = SqliteAdapter(db_path="data/databases/default/app.db") adapter.connect()

Workflow operations

workflow = adapter.create_workflow({ "id": "workflow_001", "database_name": "default", "name": "Research Pipeline", ... })

adapter.disconnect()

Or use as context manager

with SqliteAdapter(db_path="data/app.db") as adapter: workflows = adapter.list_workflows("default")

class SqliteAdapter​

SQLite storage adapter implementing all storage protocols.

Provides complete CRUD operations for all ChaosCypher entities using SQLite + SQLModel.

Architecture:

  • Entity conversion via SqliteMixinBase (delegates to utils module)
  • Implements all storage protocols via mixins (ISP-compliant)
  • Uses SQLModel for ORM
  • Manages SQLite sessions and transactions

Source file operations are split across 4 focused mixins:

  • SourceLifecycleMixin: File CRUD (upload, get, list, delete, update)
  • SourceIndexingMixin: Status lifecycle, embeddings, extraction gating
  • SourceExtractionJobsMixin: Extraction job CRUD and status
  • SourceChunkTasksMixin: Chunk task CRUD, analytics, recovery

Source operations are split across 5 focused mixins:

  • SourcesMixin: Core source CRUD (get, create, update, list)
  • SourceTagsMixin: Tag CRUD and tag-to-source assignments
  • SourceChunksMixin: Document chunk CRUD, batch ops, hierarchical grouping
  • SourceCitationsMixin: Citations, stats, orphan detection, bulk clear
  • SourceDeletionMixin: Cross-mixin delete-source cascade orchestrator

Example: adapter = SqliteAdapter(db_path="data/app.db") adapter.connect() workflow = adapter.create_workflow({...}) adapter.disconnect()

Bases: WorkflowsMixin, WorkflowExecutionsMixin, ToolsMixin, SourceLifecycleMixin, SourceIndexingMixin, SourceExtractionJobsMixin, SourceChunkTasksMixin, SourceDeletionMixin, SourcesMixin, SourceTagsMixin, SourceChunksMixin, SourceCitationsMixin, SourceRecoveryEventsMixin, ChatsMixin, TriggersMixin, LLMMetricsMixin, ExtractionSubmissionsMixin, SearchRetryQueueMixin, SystemStateMixin

Methods:

connect() -> None​

Initialize connection to SQLite database.

NOTE: Tables should already be created via initialize_database() at startup. This method just creates the session for database operations.

disconnect() -> None​

Close connection to SQLite database.

transaction() -> Iterator[None]​

Group multiple repository writes into a single database transaction.

Inside this context, repository methods call _maybe_commit() which flushes to the database buffer but does not commit. The outer transaction commits at context exit, or rolls back on any exception. Supports nesting — only the outermost context actually commits / rolls back.

Example: with adapter.transaction(): adapter.source_repository.start_commit(source_id) adapter.graph_repository.delete_graph_data_by_source(source_id)

... more writes ...

All committed atomically here

Attributes:

  • database_name
  • db_path
  • session: SafeSession | None