43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.user import User
|
|
|
|
|
|
class UsersRepository:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def get_by_sub(self, authentik_sub: str) -> User | None:
|
|
stmt = select(User).where(User.authentik_sub == authentik_sub)
|
|
return self.db.scalar(stmt)
|
|
|
|
def upsert_by_sub(
|
|
self,
|
|
authentik_sub: str,
|
|
email: str | None,
|
|
display_name: str | None,
|
|
is_active: bool,
|
|
authentik_user_id: int | None = None,
|
|
) -> User:
|
|
user = self.get_by_sub(authentik_sub)
|
|
if user is None:
|
|
user = User(
|
|
authentik_sub=authentik_sub,
|
|
authentik_user_id=authentik_user_id,
|
|
email=email,
|
|
display_name=display_name,
|
|
is_active=is_active,
|
|
)
|
|
self.db.add(user)
|
|
else:
|
|
if authentik_user_id is not None:
|
|
user.authentik_user_id = authentik_user_id
|
|
user.email = email
|
|
user.display_name = display_name
|
|
user.is_active = is_active
|
|
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
return user
|