101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.internal import verify_internal_secret
|
|
from app.db.session import get_db
|
|
from app.repositories.member_organizations_repo import MemberOrganizationsRepository
|
|
from app.repositories.organizations_repo import OrganizationsRepository
|
|
from app.repositories.users_repo import UsersRepository
|
|
from app.schemas.members import MemberListResponse, MemberOrganizationsResponse, MemberSummary
|
|
from app.schemas.organizations import OrganizationListResponse, OrganizationSummary
|
|
|
|
router = APIRouter(prefix="/internal", tags=["internal"])
|
|
|
|
|
|
def _to_member_summary(member) -> MemberSummary:
|
|
return MemberSummary(
|
|
id=member.id,
|
|
authentik_sub=member.authentik_sub,
|
|
authentik_user_id=member.authentik_user_id,
|
|
email=member.email,
|
|
display_name=member.display_name,
|
|
is_active=member.is_active,
|
|
)
|
|
|
|
|
|
def _to_org_summary(org) -> OrganizationSummary:
|
|
return OrganizationSummary(
|
|
id=org.id,
|
|
org_code=org.org_code,
|
|
name=org.name,
|
|
tax_id=org.tax_id,
|
|
status=org.status,
|
|
)
|
|
|
|
|
|
@router.get("/members", response_model=MemberListResponse)
|
|
def internal_list_members(
|
|
_: None = Depends(verify_internal_secret),
|
|
db: Session = Depends(get_db),
|
|
keyword: str | None = Query(default=None),
|
|
is_active: bool | None = Query(default=None),
|
|
limit: int = Query(default=200, ge=1, le=500),
|
|
offset: int = Query(default=0, ge=0),
|
|
) -> MemberListResponse:
|
|
repo = UsersRepository(db)
|
|
items, total = repo.list(keyword=keyword, is_active=is_active, limit=limit, offset=offset)
|
|
return MemberListResponse(items=[_to_member_summary(i) for i in items], total=total, limit=limit, offset=offset)
|
|
|
|
|
|
@router.get("/members/by-sub/{authentik_sub}", response_model=MemberSummary | None)
|
|
def internal_get_member_by_sub(
|
|
authentik_sub: str,
|
|
_: None = Depends(verify_internal_secret),
|
|
db: Session = Depends(get_db),
|
|
) -> MemberSummary | None:
|
|
repo = UsersRepository(db)
|
|
member = repo.get_by_sub(authentik_sub)
|
|
return _to_member_summary(member) if member else None
|
|
|
|
|
|
@router.get("/organizations", response_model=OrganizationListResponse)
|
|
def internal_list_organizations(
|
|
_: None = Depends(verify_internal_secret),
|
|
db: Session = Depends(get_db),
|
|
keyword: str | None = Query(default=None),
|
|
status_filter: str | None = Query(default=None, alias="status"),
|
|
limit: int = Query(default=200, ge=1, le=500),
|
|
offset: int = Query(default=0, ge=0),
|
|
) -> OrganizationListResponse:
|
|
repo = OrganizationsRepository(db)
|
|
items, total = repo.list(keyword=keyword, status=status_filter, limit=limit, offset=offset)
|
|
return OrganizationListResponse(items=[_to_org_summary(i) for i in items], total=total, limit=limit, offset=offset)
|
|
|
|
|
|
@router.get("/organizations/by-code/{org_code}", response_model=OrganizationSummary | None)
|
|
def internal_get_organization_by_code(
|
|
org_code: str,
|
|
_: None = Depends(verify_internal_secret),
|
|
db: Session = Depends(get_db),
|
|
) -> OrganizationSummary | None:
|
|
repo = OrganizationsRepository(db)
|
|
org = repo.get_by_code(org_code)
|
|
return _to_org_summary(org) if org else None
|
|
|
|
|
|
@router.get("/members/{member_id}/organizations", response_model=MemberOrganizationsResponse | None)
|
|
def internal_list_member_organizations(
|
|
member_id: str,
|
|
_: None = Depends(verify_internal_secret),
|
|
db: Session = Depends(get_db),
|
|
) -> MemberOrganizationsResponse | None:
|
|
users_repo = UsersRepository(db)
|
|
link_repo = MemberOrganizationsRepository(db)
|
|
|
|
member = users_repo.get_by_id(member_id)
|
|
if not member:
|
|
return None
|
|
|
|
orgs = link_repo.list_organizations_by_member_id(member_id)
|
|
return MemberOrganizationsResponse(member=_to_member_summary(member), organizations=[_to_org_summary(o) for o in orgs])
|