44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.role import Role
|
|
from app.models.site import Site
|
|
from app.models.site_role import SiteRole
|
|
from app.models.system import System
|
|
|
|
|
|
class SiteRolesRepository:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def list_site_role_rows(self, site_id: str) -> list[tuple[SiteRole, Role, System]]:
|
|
stmt = (
|
|
select(SiteRole, Role, System)
|
|
.join(Role, Role.id == SiteRole.role_id)
|
|
.join(System, System.id == Role.system_id)
|
|
.where(SiteRole.site_id == site_id)
|
|
.order_by(System.name.asc(), Role.name.asc())
|
|
)
|
|
return list(self.db.execute(stmt).all())
|
|
|
|
def list_role_site_rows(self, role_id: str) -> list[tuple[SiteRole, Site]]:
|
|
stmt = (
|
|
select(SiteRole, Site)
|
|
.join(Site, Site.id == SiteRole.site_id)
|
|
.where(SiteRole.role_id == role_id)
|
|
.order_by(Site.display_name.asc())
|
|
)
|
|
return list(self.db.execute(stmt).all())
|
|
|
|
def set_site_roles(self, *, site_id: str, role_ids: list[str]) -> None:
|
|
self.db.execute(delete(SiteRole).where(SiteRole.site_id == site_id))
|
|
for role_id in role_ids:
|
|
self.db.add(SiteRole(site_id=site_id, role_id=role_id))
|
|
self.db.commit()
|
|
|
|
def set_role_sites(self, *, role_id: str, site_ids: list[str]) -> None:
|
|
self.db.execute(delete(SiteRole).where(SiteRole.role_id == role_id))
|
|
for site_id in site_ids:
|
|
self.db.add(SiteRole(site_id=site_id, role_id=role_id))
|
|
self.db.commit()
|