55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
"""Organization security settings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import OrgSecuritySettings
|
|
from app.schemas import OrgSecuritySettingsRead, OrgSecuritySettingsUpdate
|
|
|
|
ALLOWED_SESSION_TIMEOUTS = {15, 30, 60, 120, 480, 1440}
|
|
|
|
|
|
async def get_or_create_security_settings(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
) -> OrgSecuritySettings:
|
|
settings = await session.scalar(
|
|
select(OrgSecuritySettings).where(
|
|
OrgSecuritySettings.company_domain == company_domain
|
|
)
|
|
)
|
|
if settings is None:
|
|
settings = OrgSecuritySettings(company_domain=company_domain)
|
|
session.add(settings)
|
|
await session.flush()
|
|
return settings
|
|
|
|
|
|
async def read_security_settings(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
) -> OrgSecuritySettingsRead:
|
|
settings = await get_or_create_security_settings(session, company_domain)
|
|
return OrgSecuritySettingsRead.model_validate(settings)
|
|
|
|
|
|
async def update_security_settings(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
payload: OrgSecuritySettingsUpdate,
|
|
) -> OrgSecuritySettingsRead:
|
|
settings = await get_or_create_security_settings(session, company_domain)
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
if "session_timeout_minutes" in updates:
|
|
timeout = updates["session_timeout_minutes"]
|
|
if timeout not in ALLOWED_SESSION_TIMEOUTS:
|
|
raise ValueError(
|
|
"Session timeout must be one of: "
|
|
+ ", ".join(str(value) for value in sorted(ALLOWED_SESSION_TIMEOUTS))
|
|
)
|
|
for field, value in updates.items():
|
|
setattr(settings, field, value)
|
|
await session.flush()
|
|
return OrgSecuritySettingsRead.model_validate(settings)
|