49 lines
1.6 KiB
Python
49 lines
1.6 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
|
|
from app.services.session_policy import validate_session_timeout_minutes
|
|
|
|
|
|
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:
|
|
validate_session_timeout_minutes(updates["session_timeout_minutes"])
|
|
for field, value in updates.items():
|
|
setattr(settings, field, value)
|
|
await session.flush()
|
|
return OrgSecuritySettingsRead.model_validate(settings)
|