74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
|
import json
|
||
|
|
from typing import Iterable, Optional, Dict
|
||
|
|
import datetime
|
||
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
||
|
|
import logging
|
||
|
|
from flask import current_app
|
||
|
|
import random, string
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
db = current_app.db
|
||
|
|
|
||
|
|
|
||
|
|
class User(db.Model):
|
||
|
|
uuid: Mapped[str] = mapped_column(primary_key=True)
|
||
|
|
username: Mapped[str] = mapped_column(unique=True)
|
||
|
|
display_name: Mapped[str] = mapped_column(server_default="")
|
||
|
|
first_name: Mapped[str] = mapped_column(server_default="")
|
||
|
|
last_name: Mapped[str] = mapped_column(server_default="")
|
||
|
|
email: Mapped[str] = mapped_column(server_default="")
|
||
|
|
phone: Mapped[str] = mapped_column(server_default="")
|
||
|
|
street: Mapped[str] = mapped_column(server_default="")
|
||
|
|
number: Mapped[str] = mapped_column(server_default="")
|
||
|
|
postal: Mapped[str] = mapped_column(server_default="")
|
||
|
|
city: Mapped[str] = mapped_column(server_default="")
|
||
|
|
country: Mapped[str] = mapped_column(server_default="")
|
||
|
|
lat: Mapped[str] = mapped_column(server_default="")
|
||
|
|
lon: Mapped[str] = mapped_column(server_default="")
|
||
|
|
birthdate: Mapped[Optional[datetime.date]]
|
||
|
|
last_updated: Mapped[datetime.datetime]
|
||
|
|
include_in_views: Mapped[Optional[bool]]
|
||
|
|
carddav_key: Mapped[str]
|
||
|
|
|
||
|
|
def reset_carddav_password(self):
|
||
|
|
self.carddav_key = "".join(
|
||
|
|
random.choices(string.ascii_lowercase + string.digits, k=20)
|
||
|
|
)
|
||
|
|
current_app.radicale.create_or_update_authfile()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def create_or_update(oidc_user: Dict) -> "User":
|
||
|
|
uuid = oidc_user["sub"]
|
||
|
|
user = db.session.execute(
|
||
|
|
db.select(User).filter_by(uuid=uuid)
|
||
|
|
).scalar_one_or_none()
|
||
|
|
if not user:
|
||
|
|
logger.debug(f"User with UUID {uuid} not yet found; creating..")
|
||
|
|
user = User(uuid=uuid, last_updated=datetime.datetime.now())
|
||
|
|
user.reset_carddav_password() # init carddav password
|
||
|
|
db.session.add(user)
|
||
|
|
user.display_name = oidc_user["name"]
|
||
|
|
if user.username != oidc_user["preferred_username"]:
|
||
|
|
# possibly changed username
|
||
|
|
user.username = oidc_user["preferred_username"]
|
||
|
|
current_app.radicale.ensure_user_exists(user)
|
||
|
|
current_app.radicale.create_or_update_authfile()
|
||
|
|
return user
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
return self.username
|
||
|
|
|
||
|
|
def to_dict(self):
|
||
|
|
return {k: self.__getattribute__(k) for k in User.__dict__.keys()}
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def all_users() -> Iterable["User"]:
|
||
|
|
return db.session.execute(db.select(User)).scalars()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def all_viewable_users() -> Iterable["User"]:
|
||
|
|
return db.session.execute(
|
||
|
|
db.select(User).filter_by(include_in_views=True)
|
||
|
|
).scalars()
|