Add custom logger option

Set env vars via pydantic BaseModel

Author: dormant-user

README.md

@@ -100,8 +100,8 @@ Licensed under the [MIT License][license]
 [label-actions-pypi]: https://github.com/thevickypedia/FastAPI-UI-Auth/actions/workflows/python-publish.yml/badge.svg
 [label-pypi]: https://img.shields.io/pypi/v/FastAPI-UI-Auth
 [label-pypi-format]: https://img.shields.io/pypi/format/FastAPI-UI-Auth
-[label-pypi-status]: https://img.shields.io/pypi/status/FastAPI-UI-Auth
-[label-pypi-package]: https://img.shields.io/badge/Pypi%20Package-FastAPI-UI-Auth-blue?style=for-the-badge&logo=Python
+[label-pypi-status]: https://img.shields.io/pypi/status/FastAPI_UI_Auth
+[label-pypi-package]: https://img.shields.io/badge/Pypi%20Package-FastAPI_UI_Auth-blue?style=for-the-badge&logo=Python
 [label-pyversion]: https://img.shields.io/badge/python-3.11%20%7C%203.12-blue
 [label-platform]: https://img.shields.io/badge/Platform-Linux|macOS|Windows-1f425f.svg
 [release-notes]: https://github.com/thevickypedia/FastAPI-UI-Auth/blob/main/release_notes.rst

uiauth/logger.py

@@ -0,0 +1,13 @@
+"""Default logger for FastAPI-UI-Auth package."""
+
+import logging
+import sys
+
+CUSTOM_LOGGER = logging.getLogger(__name__)
+CUSTOM_LOGGER.setLevel(logging.DEBUG)
+CONSOLE_HANDLER = logging.StreamHandler(sys.stdout)
+CONSOLE_FORMATTER = logging.Formatter(
+    fmt="%(levelname)-9s %(message)s",
+)
+CONSOLE_HANDLER.setFormatter(fmt=CONSOLE_FORMATTER)
+CUSTOM_LOGGER.addHandler(hdlr=CONSOLE_HANDLER)

uiauth/models.py

@@ -1,15 +1,66 @@
+import os
 import pathlib
 from typing import Callable, Dict, List, Optional, Type
 
 from fastapi.routing import APIRoute, APIWebSocketRoute
 from fastapi.templating import Jinja2Templates
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationInfo, field_validator
 
 from uiauth.enums import APIMethods
 
 templates = Jinja2Templates(directory=pathlib.Path(__file__).parent / "templates")
 
 
+def get_env(keys: List[str], default: Optional[str] = None) -> Optional[str]:
+    """Get environment variable value.
+
+    Args:
+        keys: List of environment variable names to check.
+        default: Default value if the environment variable is not set.
+
+    Returns:
+        Value of the environment variable or default value.
+    """
+    for key in keys:
+        if value := os.getenv(key):
+            return value
+        if value := os.getenv(key.upper()):
+            return value
+        if value := os.getenv(key.lower()):
+            return value
+    return default
+
+
+class EnvConfig(BaseModel):
+    """Configuration for environment variables."""
+
+    username: str
+    password: str
+
+    # noinspection PyMethodParameters
+    @field_validator("username", "password", mode="before")
+    def load_user(cls, key: str, field: ValidationInfo) -> str | None:
+        """Load environment variables into the configuration.
+
+        Args:
+            key: Environment variable key to check.
+            field: Field information for validation.
+
+        See Also:
+            - This method checks if the environment variable is set and returns its value.
+            - If the key is not set, it attempts to get the value from the environment using a helper function.
+
+        Returns:
+            str | None:
+            Value of the environment variable or None if not set.
+        """
+        if not key:
+            return get_env([field.field_name, field.field_name[:4]])
+
+
+env = EnvConfig
+
+
 class Parameters(BaseModel):
     """Parameters for the Authenticator class.
 

uiauth/service.py

@@ -1,5 +1,4 @@
 import logging
-import os
 from threading import Timer
 from typing import Dict, List
 
@@ -13,10 +12,9 @@
 from fastapi.routing import APIRoute, APIWebSocketRoute
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
-from uiauth import endpoints, enums, models, utils
+from uiauth import endpoints, enums, logger, models, utils
 
 dotenv.load_dotenv(dotenv_path=dotenv.find_dotenv(), override=True)
-LOGGER = logging.getLogger("uvicorn.default")
 BEARER_AUTH = HTTPBearer()
 
 
@@ -33,10 +31,11 @@ def __init__(
         app: FastAPI,
         params: models.Parameters | List[models.Parameters],
         timeout: int = 300,
-        username: str = os.environ.get("USERNAME"),
-        password: str = os.environ.get("PASSWORD"),
+        username: str = None,
+        password: str = None,
         fallback_button: str = models.fallback.button,
         fallback_path: str = models.fallback.path,
+        custom_logger: logging.Logger = None,
     ):
         """Initialize the APIAuthenticator with the FastAPI app and secure function.
 
@@ -48,8 +47,9 @@ def __init__(
             password: Password for authentication, can be set via environment variable 'PASSWORD'.
             fallback_button: Title for the fallback button, defaults to "LOGIN".
             fallback_path: Fallback path to redirect to in case of session timeout or invalid session.
+            custom_logger: Custom logger instance, defaults to the custom logger.
         """
-        assert all((username, password)), "'username' and 'password' are mandatory."
+        models.env = models.EnvConfig(username=username, password=password)
         assert fallback_path.startswith("/"), "Fallback path must start with '/'"
 
         self.app = app
@@ -68,11 +68,15 @@ def __init__(
             handler=utils.redirect_exception_handler,
         )
 
-        self.username = username
-        self.password = password
+        if custom_logger:
+            assert isinstance(
+                custom_logger, logging.Logger
+            ), "Custom logger must be an instance of logging.Logger"
+            logger.CUSTOM_LOGGER = custom_logger
         self.timeout = timeout
 
         self._secure()
+        logger.CUSTOM_LOGGER.debug("Endpoints registered: %s", len(self.params))
 
     def _verify_auth(
         self,
@@ -94,11 +98,11 @@ def _verify_auth(
         session_token = utils.verify_login(
             authorization=authorization,
             request=request,
-            env_username=self.username,
-            env_password=self.password,
         )
         if destination := request.cookies.get("X-Requested-By"):
-            LOGGER.info("Setting session timeout for %s seconds", self.timeout)
+            logger.CUSTOM_LOGGER.info(
+                "Setting session timeout for %s seconds", self.timeout
+            )
             # Set session_token cookie with a timeout, to be used for session validation when redirected
             response.set_cookie(
                 key="session_token",

uiauth/templates/index.html

@@ -3,7 +3,7 @@
 <html lang="en">
 <head>
     <meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-    <title>FastAPI - Authenticator</title>
+    <title>FastAPI UI Authentication</title>
     <meta property="og:type" content="Authenticator">
     <meta name="keywords" content="Python, fastapi, JavaScript, HTML, CSS">
     <meta name="author" content="Vignesh Rao">

uiauth/templates/session.html

@@ -2,7 +2,7 @@
 <html lang="en">
 <head>
     <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
-    <title>FastAPI - Authenticator</title>
+    <title>FastAPI UI Authentication</title>
     <meta property="og:type" content="Authenticator">
     <meta name="keywords" content="Python, fastapi, JavaScript, HTML, CSS">
     <meta name="author" content="Vignesh Rao">

uiauth/templates/unauthorized.html

@@ -2,7 +2,7 @@
 <html lang="en">
 <head>
     <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
-    <title>FastAPI - Authenticator</title>
+    <title>FastAPI UI Authentication</title>
     <meta property="og:type" content="Authenticator">
     <meta name="keywords" content="Python, fastapi, JavaScript, HTML, CSS">
     <meta name="author" content="Vignesh Rao">

uiauth/utils.py

@@ -1,4 +1,3 @@
-import logging
 import secrets
 from typing import List, NoReturn
 
@@ -9,9 +8,7 @@
 from fastapi.security import HTTPAuthorizationCredentials
 from fastapi.websockets import WebSocket
 
-from uiauth import enums, models, secure
-
-LOGGER = logging.getLogger("uvicorn.default")
+from uiauth import enums, logger, models, secure
 
 
 def failed_auth_counter(request: Request) -> None:
@@ -62,7 +59,7 @@ def raise_error(request: Request) -> NoReturn:
         request: Request object containing client information.
     """
     failed_auth_counter(request)
-    LOGGER.error(
+    logger.CUSTOM_LOGGER.error(
         "Incorrect username or password: %d",
         models.ws_session.invalid[request.client.host],
     )
@@ -88,16 +85,12 @@ def extract_credentials(authorization: HTTPAuthorizationCredentials) -> List[str
 def verify_login(
     authorization: HTTPAuthorizationCredentials,
     request: Request,
-    env_username: str,
-    env_password: str,
 ) -> str | NoReturn:
     """Verifies authentication and generates session token for each user.
 
     Args:
         authorization: Authorization header from the request.
         request: Request object containing client information.
-        env_username: Environment variable for the username.
-        env_password: Environment variable for the password.
 
     Returns:
         str:
@@ -107,11 +100,11 @@ def verify_login(
         username, signature, timestamp = extract_credentials(authorization)
     else:
         raise_error(request)
-    if secrets.compare_digest(username, env_username):
-        hex_user = secure.hex_encode(env_username)
-        hex_pass = secure.hex_encode(env_password)
+    if secrets.compare_digest(username, models.env.username):
+        hex_user = secure.hex_encode(models.env.username)
+        hex_pass = secure.hex_encode(models.env.password)
     else:
-        LOGGER.warning("User '%s' not allowed", username)
+        logger.CUSTOM_LOGGER.warning("User '%s' not allowed", models.env.username)
         raise_error(request)
     message = f"{hex_user}{hex_pass}{timestamp}"
     expected_signature = secure.calculate_hash(message)
@@ -151,18 +144,18 @@ def verify_session(
         and session_token
         and secrets.compare_digest(session_token, stored_token)
     ):
-        LOGGER.info("Session is valid for host: %s", request.client.host)
+        logger.CUSTOM_LOGGER.info("Session is valid for host: %s", request.client.host)
         return
     elif not session_token:
-        LOGGER.warning(
+        logger.CUSTOM_LOGGER.warning(
             "Session is invalid or expired for host: %s", request.client.host
         )
         raise models.RedirectException(
             source=request.url.path,
             destination=enums.APIEndpoints.fastapi_login,
         )
     else:
-        LOGGER.warning(
+        logger.CUSTOM_LOGGER.warning(
             "Session token mismatch for host: %s. Expected: %s, Received: %s",
             request.client.host,
             stored_token,
@@ -198,6 +191,6 @@ def clear_session(host: str) -> None:
     """
     if models.ws_session.client_auth.get(host):
         models.ws_session.client_auth.pop(host)
-        LOGGER.info("Session cleared for host: %s", host)
+        logger.CUSTOM_LOGGER.info("Session cleared for host: %s", host)
     else:
-        LOGGER.warning("No session found for host: %s", host)
+        logger.CUSTOM_LOGGER.warning("No session found for host: %s", host)