126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
"""Organization repository for database operations."""
|
|
|
|
from uuid import UUID
|
|
|
|
import asyncpg
|
|
|
|
|
|
class OrgRepository:
|
|
"""Database operations for organizations."""
|
|
|
|
def __init__(self, conn: asyncpg.Connection) -> None:
|
|
self.conn = conn
|
|
|
|
async def create(
|
|
self,
|
|
org_id: UUID,
|
|
name: str,
|
|
slug: str,
|
|
) -> dict:
|
|
"""Create a new organization."""
|
|
row = await self.conn.fetchrow(
|
|
"""
|
|
INSERT INTO orgs (id, name, slug)
|
|
VALUES ($1, $2, $3)
|
|
RETURNING id, name, slug, created_at
|
|
""",
|
|
org_id,
|
|
name,
|
|
slug,
|
|
)
|
|
return dict(row)
|
|
|
|
async def get_by_id(self, org_id: UUID) -> dict | None:
|
|
"""Get organization by ID."""
|
|
row = await self.conn.fetchrow(
|
|
"""
|
|
SELECT id, name, slug, created_at
|
|
FROM orgs
|
|
WHERE id = $1
|
|
""",
|
|
org_id,
|
|
)
|
|
return dict(row) if row else None
|
|
|
|
async def get_by_slug(self, slug: str) -> dict | None:
|
|
"""Get organization by slug."""
|
|
row = await self.conn.fetchrow(
|
|
"""
|
|
SELECT id, name, slug, created_at
|
|
FROM orgs
|
|
WHERE slug = $1
|
|
""",
|
|
slug,
|
|
)
|
|
return dict(row) if row else None
|
|
|
|
async def add_member(
|
|
self,
|
|
member_id: UUID,
|
|
user_id: UUID,
|
|
org_id: UUID,
|
|
role: str,
|
|
) -> dict:
|
|
"""Add a member to an organization."""
|
|
row = await self.conn.fetchrow(
|
|
"""
|
|
INSERT INTO org_members (id, user_id, org_id, role)
|
|
VALUES ($1, $2, $3, $4)
|
|
RETURNING id, user_id, org_id, role, created_at
|
|
""",
|
|
member_id,
|
|
user_id,
|
|
org_id,
|
|
role,
|
|
)
|
|
return dict(row)
|
|
|
|
async def get_member(self, user_id: UUID, org_id: UUID) -> dict | None:
|
|
"""Get membership for a user in an organization."""
|
|
row = await self.conn.fetchrow(
|
|
"""
|
|
SELECT om.id, om.user_id, om.org_id, om.role, om.created_at
|
|
FROM org_members om
|
|
WHERE om.user_id = $1 AND om.org_id = $2
|
|
""",
|
|
user_id,
|
|
org_id,
|
|
)
|
|
return dict(row) if row else None
|
|
|
|
async def get_members(self, org_id: UUID) -> list[dict]:
|
|
"""Get all members of an organization."""
|
|
rows = await self.conn.fetch(
|
|
"""
|
|
SELECT om.id, om.user_id, u.email, om.role, om.created_at
|
|
FROM org_members om
|
|
JOIN users u ON u.id = om.user_id
|
|
WHERE om.org_id = $1
|
|
ORDER BY om.created_at
|
|
""",
|
|
org_id,
|
|
)
|
|
return [dict(row) for row in rows]
|
|
|
|
async def get_user_orgs(self, user_id: UUID) -> list[dict]:
|
|
"""Get all organizations a user belongs to."""
|
|
rows = await self.conn.fetch(
|
|
"""
|
|
SELECT o.id, o.name, o.slug, o.created_at, om.role
|
|
FROM orgs o
|
|
JOIN org_members om ON om.org_id = o.id
|
|
WHERE om.user_id = $1
|
|
ORDER BY o.created_at
|
|
""",
|
|
user_id,
|
|
)
|
|
return [dict(row) for row in rows]
|
|
|
|
async def slug_exists(self, slug: str) -> bool:
|
|
"""Check if organization slug exists."""
|
|
result = await self.conn.fetchval(
|
|
"SELECT EXISTS(SELECT 1 FROM orgs WHERE slug = $1)",
|
|
slug,
|
|
)
|
|
return result
|