oauth_authentication.py

Go to the documentation of this file.

r"""
# +==== BEGIN CatFeeder =================+
# LOGO:
# ..............(..../\
# ...............)..(.')
# ..............(../..)
# ...............\‍(__)|
# Inspired by Joan Stark
# source https://www.asciiart.eu/
# animals/cats
# /STOP
# PROJECT: CatFeeder
# FILE: oauth_authentication.py
# CREATION DATE: 11-10-2025
# LAST Modified: 14:53:3 19-12-2025
# DESCRIPTION:
# This is the backend server in charge of making the actual website work.
# /STOP
# COPYRIGHT: (c) Cat Feeder
# PURPOSE: The file that contains all the methods for the OAuth authentication.
# // AR
# +==== END CatFeeder =================+
"""
 
 
import uuid
from datetime import datetime, timedelta
from typing import Union, Dict
import requests
from fastapi import Response, Request
from display_tty import Disp, initialise_logger
from . import constants as CONST
from ..server_header import ServerHeaders
from ..sql import SQL
from ..core import RuntimeManager, RI
from ..http_codes import HCI, HTTP_DEFAULT_TYPE
from ..boilerplates import BoilerplateIncoming, BoilerplateNonHTTP, BoilerplateResponses
 
 
class OAuthAuthentication:
    """
    The class that handle the oauth authentication
    """
 
    disp: Disp = initialise_logger(__qualname__, False)
 
    def __init__(self, success: int = 0, error: int = 84, debug: bool = False) -> None:
        """_summary_
            The constructor of the OAuth authentication class
 
        Args:
            runtime_data (RuntimeData): _description_: The initialised version of the runtime_data class.
            success (int, optional): _description_. Defaults to 0.: The status code for success.
            error (int, optional): _description_. Defaults to 84.: The status code for error.
            debug (bool, optional): _description_. Defaults to False.: The info on if to activate debug mode.
        """
        # ------------------------ The logging function ------------------------
        self.disp.update_disp_debug(debug)
        self.disp.log_debug("Initialising...")
        # -------------------------- Inherited values --------------------------
        self.debug: bool = debug
        self.success: int = success
        self.error: int = error
        self.runtime_manager_initialised: RuntimeManager = RI
        # -------------------------- Shared instances --------------------------
        self.database_link: SQL = self.runtime_manager_initialised.get(SQL)
        self.boilerplate_non_http_initialised: BoilerplateNonHTTP = self.runtime_manager_initialised.get(
            BoilerplateNonHTTP
        )
        self.boilerplate_responses_initialised: BoilerplateResponses = self.runtime_manager_initialised.get(
            BoilerplateResponses)
        self.boilerplate_incoming_initialised: BoilerplateIncoming = self.runtime_manager_initialised.get(
            BoilerplateIncoming)
        self.server_headers_initialised: ServerHeaders = self.runtime_manager_initialised.get(
            ServerHeaders)
        self.disp.log_debug("Initialised")
 
    def _generate_oauth_authorization_url(self, provider: str) -> Union[int, str]:
        """
        Generate an OAuth authorization url depends on the given provider
        """
        title = "generate_oauth_authorization_url"
        retrived_provider = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'"
        )
        self.disp.log_debug(f"Retrived provider: {retrived_provider}", title)
        if isinstance(retrived_provider, int):
            self.disp.log_error("Unknown or Unsupported OAuth provider", title)
            return self.error
        base_url = retrived_provider[0]["authorisation_base_url"]
        client_id = retrived_provider[0]["client_id"]
        scope = retrived_provider[0]["provider_scope"]
        redirect_uri = CONST.REDIRECT_URI
        state = str(uuid.uuid4())
        columns = self.database_link.get_table_column_names(
            CONST.TAB_VERIFICATION)
        self.disp.log_debug(f"Columns list: {columns}", title)
        if isinstance(columns, int):
            return self.error
        columns.pop(0)
        expiration_time = self.boilerplate_non_http_initialised.set_lifespan(
            CONST.EMAIL_VERIFICATION_DELAY)
        et_str = self.database_link.datetime_to_string(
            expiration_time, False)
        self.disp.log_debug(f"Expiration time: {et_str}", et_str)
        data: list = []
        data.append("state")
        data.append(state)
        data.append(et_str)
        if self.database_link.insert_data_into_table(CONST.TAB_VERIFICATION, data, columns) == self.error:
            return self.error
        state += ":"
        state += provider
        if provider == "google":
            url = f"{base_url}?access_type=offline&client_id={client_id}&redirect_uri={redirect_uri}&prompt=consent"
        else:
            url = f"{base_url}?client_id={client_id}&redirect_uri={redirect_uri}"
        url += f"&response_type=code&scope={scope}&state={state}"
        url = url.replace(" ", "%20")
        url = url.replace(":", "%3A")
        url = url.replace("/", "%2F")
        url = url.replace("?", "%3F")
        url = url.replace("&", "%26")
        self.disp.log_debug(f"url = {url}", title)
        return url
 
    def _exchange_code_for_token(self, provider: str, code: str):
        """
        Exchange the OAuth authorization code for an access token
        """
        title = "exchange_code_for_token"
 
        retrieved_provider = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'"
        )
        if isinstance(retrieved_provider, int):
            return self.boilerplate_responses_initialised.build_response_body(
                "exchange_code_for_token",
                "Internal server error.",
                "Internal server error.",
                None,
                True
            )
        headers: dict = {}
        headers["Accept"] = "application/json"
        headers["Content-Type"] = "application/x-www-form-urlencoded"
        token_url = retrieved_provider[0]["token_grabber_base_url"]
 
        data: dict = {}
        data["client_id"] = retrieved_provider[0]["client_id"]
        data["client_secret"] = retrieved_provider[0]["client_secret"]
        data["code"] = code
        data["redirect_uri"] = CONST.REDIRECT_URI
        data["grant_type"] = "authorization_code"
        try:
            response = requests.post(
                token_url, data=data, headers=headers, timeout=10
            )
            self.disp.log_debug(f"Exchange response = {response}", title)
            response.raise_for_status()
            token_response = response.json()
            if "error" in token_response:
                msg = "OAuth error: "
                msg += f"{token_response['error_description']}"
                self.disp.log_error(msg, title)
                return self.boilerplate_responses_initialised.build_response_body(
                    "exchange_code_for_token",
                    "Failed to get the token",
                    f"{token_response['error']}",
                    None,
                    True
                )
            return token_response
        except requests.RequestException as e:
            self.disp.log_error(f"RequestException: {str(e)}", title)
            return self.boilerplate_responses_initialised.build_response_body(
                "exchange_code_for_token",
                "HTTP request failed.",
                f"{str(e)}",
                None,
                True
            )
        except ValueError:
            self.disp.log_error("Failed to parse response JSON.", title)
            return self.boilerplate_responses_initialised.build_response_body(
                "exchange_code_for_token",
                "Invalid JSON response from provider.",
                "Invalid JSON",
                None,
                True
            )
 
    def _get_user_info(self, provider: str, access_token: str):
        """
        Get a user information depending
        """
        title: str = "get_user_info"
        retrieved_data = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'"
        )
        self.disp.log_debug(
            f"Retrieved oauth provider data: {retrieved_data}", title)
        if isinstance(retrieved_data, int):
            return self.boilerplate_responses_initialised.build_response_body(
                "get_user_info",
                "Failed to fetch the OAuth provider information.",
                "Failed to fetch the OAuth provider information.",
                None,
                True
            )
        headers = {
            "Authorization": f"Bearer {access_token}"
        }
        user_info_url = retrieved_data[0]["user_info_base_url"]
        self.disp.log_debug(f"User info headers: {headers}", title)
        self.disp.log_debug(f"User info url: {user_info_url}", title)
        response = requests.get(user_info_url, headers=headers, timeout=10)
        if response.status_code != 200:
            return self.boilerplate_responses_initialised.build_response_body(
                "get_user_info",
                "Failed to retrieve the user email from the provider",
                response,
                None,
                True
            )
        user_info = response.json()
        self.disp.log_debug(f"User info: {user_info}", title)
        if provider == "github":
            for _, info in enumerate(user_info):
                if info["primary"]:
                    email: dict = {}
                    email["email"] = info["email"]
                    return email
        return user_info
 
    def _oauth_user_logger(self, user_info: Dict, provider: str, connection_data: list) -> Response:
        """
        The function to insert or update the user information in the database
        """
        title: str = "oauth_user_logger"
        email: str = user_info["email"]
        retrieved_user = self.database_link.get_data_from_table(
            CONST.TAB_ACCOUNTS,
            "*",
            f"email='{email}'"
        )
        self.disp.log_debug(f"Retrieved user: {retrieved_user}", title)
        if isinstance(retrieved_user, int) is False:
            retrieved_provider = self.database_link.get_data_from_table(
                CONST.TAB_SERVICES,
                "*",
                f"name='{provider}'"
            )
            msg = "Retrieved provider: "
            msg += f"{retrieved_provider}"
            self.disp.log_debug(msg, title)
            if isinstance(retrieved_user, int):
                return self.boilerplate_responses_initialised.internal_server_error(title, None)
            connection_data.append(str(retrieved_provider[0]["id"]))
            connection_data.append(str(retrieved_user[0]["id"]))
            self.disp.log_debug(f"Connection data: {connection_data}", title)
 
            provider_id = str(retrieved_provider[0]["id"])
            user_id = str(retrieved_user[0]["id"])
            if isinstance(self.database_link.get_data_from_table(
                CONST.TAB_ACTIVE_OAUTHS,
                "*",
                f"service_id='{provider_id}' AND user_id='{user_id}'"
            ), int):
                columns = self.database_link.get_table_column_names(
                    CONST.TAB_ACTIVE_OAUTHS)
                if isinstance(columns, int):
                    return self.boilerplate_responses_initialised.internal_server_error(title, None)
                columns.pop(0)
                self.disp.log_debug(f"Columns list = {columns}", title)
                if self.database_link.insert_data_into_table(
                    CONST.TAB_ACTIVE_OAUTHS,
                    connection_data,
                    columns
                ) == self.error:
                    return self.boilerplate_responses_initialised.internal_server_error(title, None)
            user_data = self.boilerplate_incoming_initialised.log_user_in(
                email
            )
            if user_data["status"] == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, None)
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="User successfully logged in.",
                resp="success.",
                token=user_data["token"],
                error=False
            )
            body["token"] = user_data["token"]
            return HCI.accepted(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        columns = self.database_link.get_table_column_names(
            CONST.TAB_ACCOUNTS)
        if isinstance(columns, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        columns.pop(0)
        self.disp.log_debug(f"Columns list = {columns}", title)
        username: str = email.split('@')[0]
        user_data: list = []
        user_data.append(username)
        user_data.append(email)
        user_data.append(provider)
        user_data.append(provider)
        user_data.append("NULL")
        user_data.append(str(int(False)))
        self.disp.log_debug(f"Data list = {user_data}", title)
        if self.database_link.insert_data_into_table(CONST.TAB_ACCOUNTS, user_data, columns) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        retrieved_user = self.database_link.get_data_from_table(
            CONST.TAB_ACCOUNTS,
            "*",
            f"email='{email}'"
        )
        self.disp.log_debug(f"Retrieved user: {retrieved_user}", title)
        if isinstance(retrieved_user, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        retrieved_provider = self.database_link.get_data_from_table(
            CONST.TAB_SERVICES,
            "*",
            f"name='{provider}'"
        )
        self.disp.log_debug(f"Retrieved provider: {retrieved_provider}", title)
        if isinstance(retrieved_user, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        connection_data.append(str(retrieved_provider[0]["id"]))
        connection_data.append(str(retrieved_user[0]["id"]))
        self.disp.log_debug(f"Connection data: {connection_data}", title)
        columns = self.database_link.get_table_column_names(
            CONST.TAB_ACTIVE_OAUTHS)
        if isinstance(columns, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        columns.pop(0)
        self.disp.log_debug(f"Columns list = {columns}", title)
        if self.database_link.insert_data_into_table(
            CONST.TAB_ACTIVE_OAUTHS,
            connection_data,
            columns
        ) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        user_data = self.boilerplate_incoming_initialised.log_user_in(
            email)
        if user_data["status"] == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        body = self.boilerplate_responses_initialised.build_response_body(
            title=title,
            message="User successfully logged in.",
            resp="success.",
            token=user_data["token"],
            error=False
        )
        body["token"] = user_data["token"]
        return HCI.accepted(
            body,
            content_type=HTTP_DEFAULT_TYPE,
            headers=self.server_headers_initialised.for_json()
        )
 
    def _handle_token_response(self, token_response: Dict, provider: str) -> Response:
        """
        The function that handle the response given by the provider for the oauth token
        """
        title = "handle_token_response"
        data: list = []
        access_token: str = token_response["access_token"]
        if not access_token:
            return self.boilerplate_responses_initialised.no_access_token(title, None)
        data.append(access_token)
        self.disp.log_debug(f"Gotten access token: {access_token}", title)
        if provider == "github":
            data.append(
                self.database_link.datetime_to_string(
                    datetime.now()
                )
            )
            data.append("0")
            data.append("NULL")
        else:
            expires: int = token_response["expires_in"]
            if not expires:
                body = self.boilerplate_responses_initialised.build_response_body(
                    title,
                    "The expiration time was not found in the provided response.",
                    "Expiration time not found.",
                    token=access_token,
                    error=True
                )
                return HCI.bad_request(
                    body,
                    content_type=HTTP_DEFAULT_TYPE,
                    headers=self.server_headers_initialised.for_json()
                )
            current_time = datetime.now()
            new_time = current_time + timedelta(seconds=expires)
            expiration_date = self.database_link.datetime_to_string(
                new_time
            )
            data.append(expiration_date)
            data.append(str(expires))
            refresh_link = token_response["refresh_token"]
            if not refresh_link:
                body = self.boilerplate_responses_initialised.build_response_body(
                    title,
                    "The refresh link was not found in the provided response.",
                    "Refresh link not found.",
                    token=access_token,
                    error=True
                )
                return HCI.bad_request(
                    body,
                    content_type=HTTP_DEFAULT_TYPE,
                    headers=self.server_headers_initialised.for_json()
                )
            data.append(refresh_link)
        self.disp.log_debug(
            f"Generated data for new oauth connexion user: {data}", title)
        user_info = self._get_user_info(provider, access_token)
        if "error" in user_info:
            body = self.boilerplate_responses_initialised.build_response_body(
                title,
                user_info["error"],
                "internal error",
                token=access_token,
                error=True
            )
            return HCI.internal_server_error(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        return self._oauth_user_logger(user_info, provider, data)
 
    def refresh_token(self, provider_name: str, refresh_link: str) -> Union[str, None]:
        """
        The function that use the given provider name and refresh link to generate a new token for oauth authentication
        """
        title: str = "refresh_token"
        if any(word in provider_name for word in ("google", "discord", "spotify")):
            retrieved_data = self.database_link.get_data_from_table(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "*",
                f"provider_name='{provider_name}'"
            )
            msg = "Retrieved provider data:"
            msg += f"{retrieved_data}"
            self.disp.log_debug(msg, title)
            if isinstance(retrieved_data, int):
                self.disp.log_error(
                    "An error has been detected when retrieving the provider data", title
                )
                return None
            token_url: str = retrieved_data[0]["token_grabber_base_url"]
            generated_data: dict = {}
            generated_data["client_id"] = retrieved_data[0]["client_id"]
            generated_data["client_secret"] = retrieved_data[0]["client_secret"]
            generated_data["refresh_token"] = refresh_link
            generated_data["grant_type"] = "refresh_token"
            self.disp.log_debug(f"Generated data: {generated_data}", title)
            provider_response: Response = requests.post(
                token_url, data=generated_data, timeout=10
            )
            self.disp.log_debug(
                f"Provider response: {provider_response}", title)
            if provider_response.status_code == 200:
                token_response = provider_response.json()
                msg = "Provider response to json: "
                msg += f"{token_response}"
                self.disp.log_debug(msg, title)
                if "access_token" in token_response:
                    return token_response["access_token"]
            else:
                return None
        self.disp.log_error("The provider is not recognised", title)
        return None
 
    async def oauth_callback(self, request: Request) -> Response:
        """
        Callback of the OAuth login
        """
        title = "oauth_callback"
        query_params = request.query_params
        if not query_params:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="Query parameters not provided.",
                resp="no query parameters",
                token=None,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        self.disp.log_debug(f"Query params: {query_params}", title)
        code = query_params.get("code")
        self.disp.log_debug(f"Code: {code}", title)
        state = query_params.get("state")
        self.disp.log_debug(f"State: {state}", title)
        if not code or not state:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="Authorization code or state not provided.",
                resp="no code or state",
                token=None,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        uuid_gotten, provider = state.split(":")
        if not uuid_gotten or not provider:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="The state is in bad format.",
                resp="bad state format",
                token=None,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        self.disp.log_debug(f"Uuid retrived: {uuid_gotten}", title)
        self.disp.log_debug(f"Provider: {provider}", title)
        data = self.database_link.get_data_from_table(
            CONST.TAB_VERIFICATION,
            "*",
            f"definition='{uuid_gotten}'"
        )
        self.disp.log_debug(f"Data received: {data}", title)
        if isinstance(data, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        if isinstance(self.database_link.drop_data_from_table(
            CONST.TAB_VERIFICATION,
            f"definition='{uuid_gotten}'"
        ), int) is False:
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        token_response = self._exchange_code_for_token(provider, code)
        self.disp.log_debug(f"Token response: {token_response}", title)
        if "error" in token_response:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="Failed to get the token.",
                resp=token_response["error"],
                token=None,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        return self._handle_token_response(token_response, provider)
 
    async def oauth_login(self, request: Request) -> Response:
        """
        Get the authorization url for the OAuth login depending on the provider
        """
        title = "oauth_login"
        request_body = await self.boilerplate_incoming_initialised.get_body(request)
        self.disp.log_debug(f"Request body: {request_body}", title)
        if not request_body or "provider" not in request_body:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="The provider is not found in the request.",
                resp="no provider",
                token=None,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        provider = request_body["provider"]
        self.disp.log_debug(f"Oauth login provider: {provider}", title)
        authorization_url = self._generate_oauth_authorization_url(provider)
        self.disp.log_debug(f"Authorization url: {authorization_url}", title)
        if isinstance(authorization_url, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        body = self.boilerplate_responses_initialised.build_response_body(
            title=title,
            message="Authorization url successfully generated.",
            resp="success",
            token=None,
            error=False
        )
        body["authorization_url"] = authorization_url
        return HCI.success(
            body,
            content_type=HTTP_DEFAULT_TYPE,
            headers=self.server_headers_initialised.for_json()
        )
 
    async def add_oauth_provider(self, request: Request, provider: str = "") -> Response:
        """
        Add a new oauth provider to the database (only for admin)
        """
        title: str = "add_oauth_provider"
        if not provider:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="The provider is not provided.",
                resp="no provider",
                token=None,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        self.disp.log_debug(f"Provider: {provider}", title)
        token: str = self.boilerplate_incoming_initialised.get_token_if_present(
            request
        )
        if self.boilerplate_non_http_initialised.is_token_admin(token) is False:
            self.disp.log_error("You're not admin.", title)
            return self.boilerplate_responses_initialised.insuffisant_rights(title, token)
        retrived_data = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'"
        )
        if isinstance(retrived_data, int) is False:
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="The provider already exist in the database.",
                resp="provider already exist",
                token=token,
                error=True
            )
            return HCI.conflict(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        request_body = await self.boilerplate_incoming_initialised.get_body(request)
        if not request_body or not all(key in request_body for key in ("client_id", "client_secret", "provider_scope", "authorisation_base_url", "token_grabber_base_url", "user_info_base_url")):
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="A variable is missing in the body.",
                resp="missing variable",
                token=token,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        self.disp.log_debug(f"Request body: {request_body}", title)
        client_id = request_body["client_id"]
        client_secret = request_body["client_secret"]
        provider_scope = request_body["provider_scope"]
        authorisation_base_url = request_body["authorisation_base_url"]
        token_grabber_base_url = request_body["token_grabber_base_url"]
        user_info_base_url = request_body["user_info_base_url"]
        data: list = [
            provider,
            client_id,
            client_secret,
            provider_scope,
            authorisation_base_url,
            token_grabber_base_url,
            user_info_base_url
        ]
        columns = self.database_link.get_table_column_names(
            CONST.TAB_USER_OAUTH_CONNECTION
        )
        if isinstance(columns, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, token)
        columns.pop(0)
        self.disp.log_debug(f"Columns: {columns}", title)
        if self.database_link.insert_data_into_table(CONST.TAB_USER_OAUTH_CONNECTION, data, columns) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(title, token)
        oauth: list = ["1"]
        column: list = ["oauth"]
        if self.database_link.update_data_in_table(
            CONST.TAB_SERVICES,
            oauth,
            column,
            f"name='{provider}'"
        ) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(title, token)
        body = self.boilerplate_responses_initialised.build_response_body(
            title=title,
            message="The provider is successfully added.",
            resp="success",
            token=token,
            error=False
        )
        return HCI.success(
            body,
            content_type=HTTP_DEFAULT_TYPE,
            headers=self.server_headers_initialised.for_json()
        )
 
    async def update_oauth_provider_data(self, request: Request, provider: str) -> Response:
        """
        The function that modify every value of an oauth provider
        """
        title: str = "update_oauth_provider_data"
        if not provider:
            return self.boilerplate_responses_initialised.provider_not_given(title, None)
        self.disp.log_debug(f"Provider: {provider}", title)
        token: str = self.boilerplate_incoming_initialised.get_token_if_present(
            request
        )
        self.disp.log_debug(f"Token received: {token}", title)
        if self.boilerplate_non_http_initialised.is_token_admin(token) is False:
            self.disp.log_error("You're not admin.", title)
            return self.boilerplate_responses_initialised.insuffisant_rights(title, token)
        retrived_data = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'")
        if isinstance(retrived_data, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        request_body = await self.boilerplate_incoming_initialised.get_body(request)
        if not request_body or not all(key in request_body for key in ("client_id", "client_secret", "provider_scope", "authorisation_base_url", "token_grabber_base_url", "user_info_base_url")):
            body = self.boilerplate_responses_initialised.build_response_body(
                title=title,
                message="A variable is missing in the body.",
                resp="missing variable",
                token=token,
                error=True
            )
            return HCI.bad_request(
                body,
                content_type=HTTP_DEFAULT_TYPE,
                headers=self.server_headers_initialised.for_json()
            )
        self.disp.log_debug(f"Request body: {request_body}", title)
        client_id = request_body["client_id"]
        client_secret = request_body["client_secret"]
        provider_scope = request_body["provider_scope"]
        authorisation_base_url = request_body["authorisation_base_url"]
        token_grabber_base_url = request_body["token_grabber_base_url"]
        user_info_base_url = request_body["user_info_base_url"]
        data: list = [
            client_id,
            client_secret,
            provider_scope,
            authorisation_base_url,
            token_grabber_base_url,
            user_info_base_url
        ]
        self.disp.log_debug(f"Generated data: {data}", title)
        columns: list = self.database_link.get_table_column_names(
            CONST.TAB_USER_OAUTH_CONNECTION
        )
        if isinstance(columns, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        columns.pop(0)
        columns.pop(0)
        self.disp.log_debug(f"Columns: {columns}", title)
        if self.database_link.update_data_in_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            data,
            columns,
            f"provider_name='{provider}'"
        ) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(title, None)
        body = self.boilerplate_responses_initialised.build_response_body(
            title=title,
            message="The provider is successfully updated.",
            resp="success",
            token=token,
            error=False
        )
        return HCI.success(
            body,
            content_type=HTTP_DEFAULT_TYPE,
            headers=self.server_headers_initialised.for_json()
        )
 
    async def patch_oauth_provider_data(self, request: Request, provider: str) -> Response:
        """
        The function that modify every value of an oauth provider
        """
        title: str = "update_oauth_provider_data"
        if not provider:
            return self.boilerplate_responses_initialised.provider_not_given(title, None)
        self.disp.log_debug(f"Provider: {provider}", title)
        token: str = self.boilerplate_incoming_initialised.get_token_if_present(
            request
        )
        self.disp.log_debug(f"Token gotten: {token}", title)
        if self.boilerplate_non_http_initialised.is_token_admin(token) is False:
            self.disp.log_error("You're not admin.", title)
            return self.boilerplate_responses_initialised.insuffisant_rights(title, token)
        retrived_data = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'"
        )
        if isinstance(retrived_data, int):
            return self.boilerplate_responses_initialised.internal_server_error(title, token)
        request_body = await self.boilerplate_incoming_initialised.get_body(request)
        if not request_body:
            return self.boilerplate_responses_initialised.missing_variable_in_body(title, token)
        self.disp.log_debug(f"Request body: {request_body}", title)
        if "provider_name" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "provider_name",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        if "client_id" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "client_id",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        if "client_secret" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "client_secret",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        if "provider_scope" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "provider_scope",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        if "authorisation_base_url" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "authorisation_base_url",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        if "token_grabber_base_url" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "token_grabber_base_url",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        if "user_info_base_url" in request_body:
            if self.boilerplate_non_http_initialised.update_single_data(
                CONST.TAB_USER_OAUTH_CONNECTION,
                "provider_name",
                "user_info_base_url",
                provider,
                request_body
            ) == self.error:
                return self.boilerplate_responses_initialised.internal_server_error(title, token)
        body = self.boilerplate_responses_initialised.build_response_body(
            title=title,
            message="The provider is successfully updated.",
            resp="success",
            token=token,
            error=False
        )
        return HCI.success(body, content_type=HTTP_DEFAULT_TYPE, headers=self.server_headers_initialised.for_json())
 
    async def delete_oauth_provider(self, request: Request, provider: str) -> Response:
        """
        The function to delete an oauth provider from the database
        """
        title: str = "delete_oauth_provider"
        if not provider:
            return self.boilerplate_responses_initialised.provider_not_given(
                title,
                None
            )
        self.disp.log_debug(f"Provider: {provider}", title)
        token: str = self.boilerplate_incoming_initialised.get_token_if_present(
            request
        )
        self.disp.log_debug(f"Token gotten: {token}", title)
        if self.boilerplate_non_http_initialised.is_token_admin(token) is False:
            self.disp.log_error("You're not admin.", title)
            return self.boilerplate_responses_initialised.insuffisant_rights(
                title,
                token
            )
        retrived_data = self.database_link.get_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            "*",
            f"provider_name='{provider}'"
        )
        if isinstance(retrived_data, int):
            return self.boilerplate_responses_initialised.internal_server_error(
                title,
                token
            )
        retrieved_service_id = self.database_link.get_data_from_table(
            CONST.TAB_SERVICES,
            "id",
            f"name='{provider}'"
        )
        if isinstance(retrieved_service_id, int):
            return self.boilerplate_responses_initialised.internal_server_error(
                title,
                token
            )
        service_id = str(retrieved_service_id[0]["id"])
        if self.database_link.drop_data_from_table(
            CONST.TAB_ACTIVE_OAUTHS,
            f"service_id='{service_id}'"
        ) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(
                title,
                token
            )
        if self.database_link.update_data_in_table(
            CONST.TAB_SERVICES,
            "0",
            "oauth",
            f"id='{service_id}'"
        ) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(
                title,
                token
            )
        if self.database_link.drop_data_from_table(
            CONST.TAB_USER_OAUTH_CONNECTION,
            f"provider_name='{provider}'"
        ) == self.error:
            return self.boilerplate_responses_initialised.internal_server_error(
                title,
                token
            )
        body = self.boilerplate_responses_initialised.build_response_body(
            title=title,
            message="The oauth provider has been deleted successfully.",
            resp="success",
            token=token,
        )
        return HCI.success(
            body,
            content_type=HTTP_DEFAULT_TYPE,
            headers=self.server_headers_initialised.for_json()
        )