Xiaoxia AI
e4e2595e70
- adapters: SQLAlchemy implementations for all 4 repositories (Project, AssetLibrary, Asset, IngestJob) - models: SQLAlchemy ORM models with proper schema - database: connection config and session management - tests: SQLAlchemy repository integration test (in-memory SQLite) - all 7 integration tests passing
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from packages.domain import AssetLibrary, AssetLibraryKind
|
|
from packages.adapters.sqlalchemy_impl.models import AssetLibraryModel
|
|
|
|
|
|
class SQLAlchemyAssetLibraryRepository:
|
|
def __init__(self, session: Session):
|
|
self.session = session
|
|
|
|
def list_by_project(self, project_id: str, kind: AssetLibraryKind | None = None) -> list[AssetLibrary]:
|
|
query = self.session.query(AssetLibraryModel).filter(AssetLibraryModel.project_id == project_id)
|
|
if kind is not None:
|
|
query = query.filter(AssetLibraryModel.kind == kind.value)
|
|
models = query.all()
|
|
return [
|
|
AssetLibrary(
|
|
id=m.id,
|
|
workspace_id=m.workspace_id,
|
|
project_id=m.project_id,
|
|
name=m.name,
|
|
kind=AssetLibraryKind(m.kind),
|
|
created_at=m.created_at,
|
|
)
|
|
for m in models
|
|
]
|
|
|
|
def create(self, library: AssetLibrary) -> AssetLibrary:
|
|
model = AssetLibraryModel(
|
|
id=library.id,
|
|
workspace_id=library.workspace_id,
|
|
project_id=library.project_id,
|
|
name=library.name,
|
|
kind=library.kind.value,
|
|
created_at=library.created_at,
|
|
)
|
|
self.session.add(model)
|
|
self.session.commit()
|
|
return library
|