78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.domain.keys import generate_experiment_key, generate_variant_key
|
|
from app.domain.mappers import to_experiment
|
|
from app.repositories.directus.experiments import ExperimentRepository
|
|
from app.repositories.directus.variants import VariantRepository
|
|
from app.schemas.admin import ExperimentCreate, ExperimentRead, ExperimentUpdate, experiment_to_read_model
|
|
|
|
|
|
class ExperimentService:
|
|
"""Application service for experiment use-cases.
|
|
|
|
This is where we will keep:
|
|
- validation
|
|
- permission checks
|
|
- state transitions
|
|
- orchestration before writing to Directus
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
repository: ExperimentRepository | None = None,
|
|
variant_repository: VariantRepository | None = None,
|
|
) -> None:
|
|
self.repository = repository or ExperimentRepository()
|
|
self.variant_repository = variant_repository or VariantRepository()
|
|
|
|
async def list_experiments(self, access_token: str | None = None) -> list[ExperimentRead]:
|
|
items = await self.repository.list(access_token=access_token)
|
|
return [experiment_to_read_model(to_experiment(item)) for item in items]
|
|
|
|
async def get_experiment(
|
|
self,
|
|
experiment_id: str,
|
|
access_token: str | None = None,
|
|
) -> ExperimentRead | None:
|
|
item = await self.repository.get(experiment_id, access_token=access_token)
|
|
if not item:
|
|
return None
|
|
return experiment_to_read_model(to_experiment(item))
|
|
|
|
async def create_experiment(
|
|
self,
|
|
payload: ExperimentCreate,
|
|
access_token: str | None = None,
|
|
) -> ExperimentRead:
|
|
data: dict[str, Any] = payload.model_dump(exclude_none=True)
|
|
data["experiment_key"] = generate_experiment_key()
|
|
item = await self.repository.create(data, access_token=access_token)
|
|
experiment = to_experiment(item)
|
|
|
|
# Auto-create 原始版本 (control) variant
|
|
await self.variant_repository.create(
|
|
{
|
|
"experiment_id": experiment.id,
|
|
"variant_key": generate_variant_key(),
|
|
"name": "原始版本",
|
|
"traffic_weight": 50,
|
|
},
|
|
access_token=access_token,
|
|
)
|
|
|
|
return experiment_to_read_model(experiment)
|
|
|
|
async def update_experiment(
|
|
self,
|
|
experiment_id: str,
|
|
payload: ExperimentUpdate,
|
|
access_token: str | None = None,
|
|
) -> ExperimentRead | None:
|
|
data: dict[str, Any] = payload.model_dump(exclude_unset=True)
|
|
item = await self.repository.update(experiment_id, data, access_token=access_token)
|
|
if not item:
|
|
return None
|
|
return experiment_to_read_model(to_experiment(item))
|