229 lines
6.4 KiB
Python
229 lines
6.4 KiB
Python
from components.models.users import (
|
|
CredentialAdd,
|
|
CredentialPatch,
|
|
Credential,
|
|
User,
|
|
UserAdd,
|
|
UserPatch,
|
|
UserProfile,
|
|
UserProfilePatch,
|
|
UserSession,
|
|
constr,
|
|
validate_call,
|
|
UUID,
|
|
)
|
|
from components.utils import merge_models
|
|
from components.database import *
|
|
from components.database import SYSTEM_CACHE_ID
|
|
from components.cache import buster
|
|
|
|
|
|
@validate_call
|
|
async def what_id(login: str):
|
|
db_params = evaluate_db_params()
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
user = db.table("users").get(Query().login == login)
|
|
|
|
if user:
|
|
return user["id"]
|
|
else:
|
|
raise ValueError("login", "The provided login name is unknown")
|
|
|
|
|
|
@validate_call
|
|
async def create(data: dict):
|
|
db_params = evaluate_db_params()
|
|
create_user = UserAdd.model_validate(data)
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
if db.table("users").search(Query().login == create_user.login):
|
|
raise ValueError("name", "The provided login name exists")
|
|
insert_data = create_user.model_dump(mode="json")
|
|
db.table("users").insert(insert_data)
|
|
|
|
return insert_data["id"]
|
|
|
|
|
|
@validate_call
|
|
async def get(user_id: UUID):
|
|
db_params = evaluate_db_params()
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
if not str(user_id) in IN_MEMORY_DB["CACHE"]["MODELS"][SYSTEM_CACHE_ID]:
|
|
IN_MEMORY_DB["CACHE"]["MODELS"][SYSTEM_CACHE_ID][
|
|
str(user_id)
|
|
] = User.model_validate(db.table("users").get(Query().id == str(user_id)))
|
|
|
|
user = IN_MEMORY_DB["CACHE"]["MODELS"][SYSTEM_CACHE_ID][str(user_id)].copy()
|
|
|
|
return user
|
|
|
|
|
|
@validate_call
|
|
async def delete(user_id: UUID):
|
|
db_params = evaluate_db_params()
|
|
user = await get(user_id=user_id)
|
|
|
|
if not user:
|
|
raise ValueError("name", "The provided user does not exist")
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
if len(db.table("users").all()) == 1:
|
|
raise ValueError("name", "Cannot delete last user")
|
|
|
|
db.table("users").remove(Query().id == str(user_id))
|
|
buster(user.id)
|
|
return user.id
|
|
|
|
|
|
@validate_call
|
|
async def create_credential(user_id: UUID, data: dict):
|
|
db_params = evaluate_db_params()
|
|
credential = CredentialAdd.model_validate(data)
|
|
user = await get(user_id=user_id)
|
|
if not user:
|
|
raise ValueError("name", "The provided user does not exist")
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
user.credentials.append(credential)
|
|
db.table("users").update(
|
|
{"credentials": user.model_dump(mode="json")["credentials"]},
|
|
Query().id == str(user_id),
|
|
)
|
|
buster(user.id)
|
|
return credential.id
|
|
|
|
|
|
@validate_call
|
|
async def delete_credential(
|
|
user_id: UUID, hex_id: constr(pattern=r"^[0-9a-fA-F]+$", min_length=2)
|
|
):
|
|
db_params = evaluate_db_params()
|
|
user = await get(user_id=user_id)
|
|
|
|
if not user:
|
|
raise ValueError("name", "The provided user does not exist")
|
|
|
|
matched_user_credential = next(
|
|
(c for c in user.credentials if c.id == bytes.fromhex(hex_id)), None
|
|
)
|
|
|
|
if not matched_user_credential:
|
|
raise ValueError(
|
|
"hex_id",
|
|
"The provided credential ID was not found in user context",
|
|
)
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
user.credentials.remove(matched_user_credential)
|
|
db.table("users").update(
|
|
{"credentials": user.model_dump(mode="json")["credentials"]},
|
|
Query().id == str(user_id),
|
|
)
|
|
buster(user_id)
|
|
return hex_id
|
|
|
|
|
|
@validate_call
|
|
async def patch(user_id: UUID, data: dict):
|
|
db_params = evaluate_db_params()
|
|
user = await get(user_id=user_id)
|
|
|
|
if not user:
|
|
raise ValueError("name", "The provided user does not exist")
|
|
|
|
patch_data = UserPatch.model_validate(data)
|
|
patched_user = merge_models(
|
|
user,
|
|
patch_data,
|
|
exclude_strategies=["exclude_override_none"],
|
|
)
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
if db.table("users").get(
|
|
(Query().login == patched_user.login) & (Query().id != str(user_id))
|
|
):
|
|
raise ValueError("login", "The provided login name exists")
|
|
|
|
db.table("users").update(
|
|
patched_user.model_dump(mode="json"),
|
|
Query().id == str(user_id),
|
|
)
|
|
buster(user.id)
|
|
return user.id
|
|
|
|
|
|
@validate_call
|
|
async def patch_profile(user_id: UUID, data: dict):
|
|
db_params = evaluate_db_params()
|
|
user = await get(user_id=user_id)
|
|
|
|
if not user:
|
|
raise ValueError("name", "The provided user does not exist")
|
|
|
|
patch_data = UserProfilePatch.model_validate(data)
|
|
patched_user_profile = merge_models(
|
|
user.profile, patch_data, exclude_strategies=["exclude_override_none"]
|
|
)
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
db.table("users").update(
|
|
{"profile": patched_user_profile.model_dump(mode="json")},
|
|
Query().id == str(user_id),
|
|
)
|
|
buster(user.id)
|
|
return user_id
|
|
|
|
|
|
@validate_call
|
|
async def patch_credential(
|
|
user_id: UUID, hex_id: constr(pattern=r"^[0-9a-fA-F]+$", min_length=2), data: dict
|
|
):
|
|
db_params = evaluate_db_params()
|
|
user = await get(user_id=user_id)
|
|
|
|
if not user:
|
|
raise ValueError("name", "The provided user does not exist")
|
|
|
|
matched_user_credential = next(
|
|
(c for c in user.credentials if c.id == bytes.fromhex(hex_id)), None
|
|
)
|
|
|
|
if not matched_user_credential:
|
|
raise ValueError(
|
|
"hex_id",
|
|
"The provided credential ID was not found in user context",
|
|
)
|
|
|
|
user.credentials.remove(matched_user_credential)
|
|
|
|
patched_credential = merge_models(
|
|
matched_user_credential,
|
|
CredentialPatch.model_validate(data),
|
|
exclude_strategies=["exclude_override_none"],
|
|
)
|
|
|
|
user.credentials.append(patched_credential)
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
db.table("users").update(
|
|
{"credentials": user.model_dump(mode="json")["credentials"]},
|
|
Query().id == str(user_id),
|
|
)
|
|
buster(user_id)
|
|
return hex_id
|
|
|
|
|
|
@validate_call
|
|
async def search(name: constr(strip_whitespace=True, min_length=0)):
|
|
db_params = evaluate_db_params()
|
|
|
|
def search_name(s):
|
|
return name in s
|
|
|
|
async with TinyDB(**db_params) as db:
|
|
matches = db.table("users").search(Query().login.test(search_name))
|
|
|
|
return [await get(user["id"]) for user in matches]
|