77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
"""Department sections within an organization."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import OrgSection
|
|
from app.schemas import OrgSectionCreate, OrgSectionRead
|
|
|
|
DEFAULT_SECTIONS = ["Admin", "HR", "Engineering", "Finance"]
|
|
|
|
|
|
async def ensure_default_sections(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
) -> None:
|
|
existing = await session.scalar(
|
|
select(OrgSection.id)
|
|
.where(OrgSection.company_domain == company_domain)
|
|
.limit(1)
|
|
)
|
|
if existing:
|
|
return
|
|
for name in DEFAULT_SECTIONS:
|
|
session.add(OrgSection(company_domain=company_domain, name=name))
|
|
await session.flush()
|
|
|
|
|
|
async def list_org_sections(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
) -> list[OrgSectionRead]:
|
|
await ensure_default_sections(session, company_domain)
|
|
result = await session.scalars(
|
|
select(OrgSection)
|
|
.where(OrgSection.company_domain == company_domain)
|
|
.order_by(OrgSection.name.asc())
|
|
)
|
|
return [OrgSectionRead.model_validate(section) for section in result.all()]
|
|
|
|
|
|
async def create_org_section(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
payload: OrgSectionCreate,
|
|
) -> OrgSectionRead:
|
|
name = payload.name.strip()
|
|
if not name:
|
|
raise ValueError("Section name is required.")
|
|
|
|
existing = await session.scalar(
|
|
select(OrgSection).where(
|
|
OrgSection.company_domain == company_domain,
|
|
OrgSection.name.ilike(name),
|
|
)
|
|
)
|
|
if existing:
|
|
raise ValueError("That section already exists.")
|
|
|
|
section = OrgSection(company_domain=company_domain, name=name)
|
|
session.add(section)
|
|
await session.flush()
|
|
return OrgSectionRead.model_validate(section)
|
|
|
|
|
|
async def get_org_section(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
section_id,
|
|
) -> OrgSection | None:
|
|
return await session.scalar(
|
|
select(OrgSection).where(
|
|
OrgSection.id == section_id,
|
|
OrgSection.company_domain == company_domain,
|
|
)
|
|
)
|