diff --git a/selfprivacy_api/repositories/users/kanidm_user_repository.py b/selfprivacy_api/repositories/users/kanidm_user_repository.py new file mode 100644 index 0000000..e5b2182 --- /dev/null +++ b/selfprivacy_api/repositories/users/kanidm_user_repository.py @@ -0,0 +1,38 @@ +from typing import Optional + +from selfprivacy_api.repositories.users.abstract_user_repository import ( + AbstractUserRepository, + UserDataUser, +) + +from selfprivacy_api.utils.kanidm_manager import ( + create_user, + delete_user, + get_user_by_username, + get_users, + update_user, +) + + +class KanidmUserRepository(AbstractUserRepository): + def get_users( + exclude_primary: bool = False, + exclude_root: bool = False, + ) -> list[UserDataUser]: + return get_users(exclude_primary=exclude_primary, exclude_root=exclude_root) + + def create_user(username: str, password: str): + """Creates a new user""" + return create_user(username=username, password=password) + + def delete_user(username: str) -> None: + """Deletes an existing user""" + return delete_user(username=username) + + def update_user(username: str, password: str) -> None: + """Updates the password of an existing user""" + return update_user(username=username, password=password) + + def get_user_by_username(username: str) -> Optional[UserDataUser]: + """Retrieves user data (UserDataUser) by username""" + return get_user_by_username(username=username) diff --git a/selfprivacy_api/utils/kanidm_manager.py b/selfprivacy_api/utils/kanidm_manager.py new file mode 100644 index 0000000..a393344 --- /dev/null +++ b/selfprivacy_api/utils/kanidm_manager.py @@ -0,0 +1,71 @@ +"""Kanidm queries.""" + +# pylint: disable=too-few-public-methods +import requests + +import strawberry + +from typing import Annotated, Union + +KANIDM_URL = "http://localhost:9001" + + +@strawberry.type +class KanidmQueryError: + error: str + + +KanidmValuesResult = Annotated[ + Union[str, KanidmQueryError], # WIP. TODO: change str + strawberry.union("KanidmValuesResult"), +] + + +# WIP WIP WIP WIP WIP WIP + + +class KanidmQueries: + @staticmethod + def _send_query(query: str) -> Union[dict, KanidmQueryError]: + try: + response = requests.get( + f"{KANIDM_URL}/api/v1/query", + params={ + "query": query, + }, + timeout=0.8, # TODO: change timeout + ) + if response.status_code != 200: + return KanidmQueryError( + error=f"Kanidm returned unexpected HTTP status code. Error: {response.text}. The query was {query}" + ) + json = response.json() + + return json["data"] + except Exception as error: + return KanidmQueryError(error=f"Kanidm request failed! Error: {str(error)}") + + @staticmethod + def create_user(username: str, password: str) -> KanidmValuesResult: + query = """""" + + data = KanidmQueries._send_query(query=query) + + if isinstance(data, KanidmQueryError): + return data + + return KanidmValuesResult(data) + + +# def get_users( +# exclude_primary: bool = False, +# exclude_root: bool = False, +# ) -> list[UserDataUser]: + +# def create_user(username: str, password: str): + +# def delete_user(username: str) -> None: + +# def update_user(username: str, password: str) -> None: + +# def get_user_by_username(username: str) -> Optional[UserDataUser]: