27 lines
1.2 KiB
Python
27 lines
1.2 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.core.errors import ConflictAgentError
|
|
from app.service.authorization_service import AuthorizationService
|
|
from app.service.config_release_service import ConfigReleaseError, ConfigReleaseService
|
|
|
|
|
|
class ConfigApplicationService:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def transition(
|
|
self, release_id: int, action: str, context: RequestContext
|
|
) -> dict[str, object]:
|
|
permission = {"submit_for_review": "config:write", "approve": "config:review",
|
|
"activate": "config:activate", "rollback": "config:activate"}[action]
|
|
await AuthorizationService.require(context, permission, admin=True)
|
|
service = ConfigReleaseService(self.session)
|
|
try:
|
|
async with self.session.begin():
|
|
release = await getattr(service, action)(release_id, int(context.user_id))
|
|
except ConfigReleaseError as exc:
|
|
raise ConflictAgentError(str(exc)) from exc
|
|
return {"data": {"release_id": str(release.id), "release_no": release.release_no,
|
|
"status": release.status}}
|