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
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
import json
|
|
from sqlalchemy.orm import Session
|
|
|
|
from packages.domain import Asset
|
|
from packages.adapters.sqlalchemy_impl.models import AssetModel
|
|
|
|
|
|
class SQLAlchemyAssetRepository:
|
|
def __init__(self, session: Session):
|
|
self.session = session
|
|
|
|
def list_by_library(self, library_id: str) -> list[Asset]:
|
|
models = self.session.query(AssetModel).filter(AssetModel.library_id == library_id).all()
|
|
return [
|
|
Asset(
|
|
id=m.id,
|
|
workspace_id=m.workspace_id,
|
|
project_id=m.project_id,
|
|
library_id=m.library_id,
|
|
name=m.name,
|
|
storage_key=m.storage_key,
|
|
mime_type=m.mime_type,
|
|
metadata=json.loads(m.metadata_json),
|
|
created_at=m.created_at,
|
|
)
|
|
for m in models
|
|
]
|
|
|
|
def create(self, asset: Asset) -> Asset:
|
|
model = AssetModel(
|
|
id=asset.id,
|
|
workspace_id=asset.workspace_id,
|
|
project_id=asset.project_id,
|
|
library_id=asset.library_id,
|
|
name=asset.name,
|
|
storage_key=asset.storage_key,
|
|
mime_type=asset.mime_type,
|
|
metadata_json=json.dumps(asset.metadata),
|
|
created_at=asset.created_at,
|
|
)
|
|
self.session.add(model)
|
|
self.session.commit()
|
|
return asset
|