81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
|
"""Service repository for database operations."""
|
||
|
|
|
||
|
|
from uuid import UUID
|
||
|
|
|
||
|
|
import asyncpg
|
||
|
|
|
||
|
|
|
||
|
|
class ServiceRepository:
|
||
|
|
"""Database operations for services."""
|
||
|
|
|
||
|
|
def __init__(self, conn: asyncpg.Connection) -> None:
|
||
|
|
self.conn = conn
|
||
|
|
|
||
|
|
async def create(
|
||
|
|
self,
|
||
|
|
service_id: UUID,
|
||
|
|
org_id: UUID,
|
||
|
|
name: str,
|
||
|
|
slug: str,
|
||
|
|
) -> dict:
|
||
|
|
"""Create a new service."""
|
||
|
|
row = await self.conn.fetchrow(
|
||
|
|
"""
|
||
|
|
INSERT INTO services (id, org_id, name, slug)
|
||
|
|
VALUES ($1, $2, $3, $4)
|
||
|
|
RETURNING id, org_id, name, slug, created_at
|
||
|
|
""",
|
||
|
|
service_id,
|
||
|
|
org_id,
|
||
|
|
name,
|
||
|
|
slug,
|
||
|
|
)
|
||
|
|
return dict(row)
|
||
|
|
|
||
|
|
async def get_by_id(self, service_id: UUID) -> dict | None:
|
||
|
|
"""Get service by ID."""
|
||
|
|
row = await self.conn.fetchrow(
|
||
|
|
"""
|
||
|
|
SELECT id, org_id, name, slug, created_at
|
||
|
|
FROM services
|
||
|
|
WHERE id = $1
|
||
|
|
""",
|
||
|
|
service_id,
|
||
|
|
)
|
||
|
|
return dict(row) if row else None
|
||
|
|
|
||
|
|
async def get_by_org(self, org_id: UUID) -> list[dict]:
|
||
|
|
"""Get all services for an organization."""
|
||
|
|
rows = await self.conn.fetch(
|
||
|
|
"""
|
||
|
|
SELECT id, org_id, name, slug, created_at
|
||
|
|
FROM services
|
||
|
|
WHERE org_id = $1
|
||
|
|
ORDER BY name
|
||
|
|
""",
|
||
|
|
org_id,
|
||
|
|
)
|
||
|
|
return [dict(row) for row in rows]
|
||
|
|
|
||
|
|
async def get_by_slug(self, org_id: UUID, slug: str) -> dict | None:
|
||
|
|
"""Get service by org and slug."""
|
||
|
|
row = await self.conn.fetchrow(
|
||
|
|
"""
|
||
|
|
SELECT id, org_id, name, slug, created_at
|
||
|
|
FROM services
|
||
|
|
WHERE org_id = $1 AND slug = $2
|
||
|
|
""",
|
||
|
|
org_id,
|
||
|
|
slug,
|
||
|
|
)
|
||
|
|
return dict(row) if row else None
|
||
|
|
|
||
|
|
async def slug_exists(self, org_id: UUID, slug: str) -> bool:
|
||
|
|
"""Check if service slug exists in organization."""
|
||
|
|
result = await self.conn.fetchval(
|
||
|
|
"SELECT EXISTS(SELECT 1 FROM services WHERE org_id = $1 AND slug = $2)",
|
||
|
|
org_id,
|
||
|
|
slug,
|
||
|
|
)
|
||
|
|
return result
|