Utilities¶

Examples of ArchiPy's utility functions:

Datetime Utils¶

Work with dates and times consistently:

import logging

from archipy.helpers.utils.datetime_utils import DatetimeUtils

# Configure logging
logger = logging.getLogger(__name__)

# Get current UTC time
now = DatetimeUtils.get_datetime_utc_now()

# Format for storage/transmission
date_str = DatetimeUtils.get_string_datetime_from_datetime(now)

# Parse date string
parsed = DatetimeUtils.get_datetime_from_string_datetime(date_str)

# Convert to Jalali (Persian) calendar
jalali_date = DatetimeUtils.convert_to_jalali(now)

# Check if date is a holiday in Iran
is_holiday = DatetimeUtils.is_holiday_in_iran(now)
logger.info(f"Is holiday: {is_holiday}")

JWT Utils¶

Generate and verify JWT tokens:

import logging
from uuid import uuid4

from archipy.helpers.utils.jwt_utils import JWTUtils
from archipy.models.errors import InvalidTokenError, TokenExpiredError

# Configure logging
logger = logging.getLogger(__name__)

# Generate a user access token
user_id = uuid4()
access_token = JWTUtils.create_access_token(user_id)

# Generate a refresh token with additional claims
additional_claims = {"user_role": "admin", "permissions": ["read", "write"]}
refresh_token = JWTUtils.create_refresh_token(user_id, additional_claims=additional_claims)

# Verify a token
try:
    payload = JWTUtils.verify_access_token(access_token)
except (InvalidTokenError, TokenExpiredError) as e:
    logger.error(f"Invalid token: {e}")
    raise
else:
    logger.info(f"Token valid for user: {payload['sub']}")

# Get token expiration time
expiry = JWTUtils.get_token_expiry(access_token)
logger.debug(f"Token expires at: {expiry}")

# Extract user UUID from token payload
user_uuid = JWTUtils.extract_user_uuid(payload)

Password Utils¶

Secure password handling:

import logging

from archipy.helpers.utils.password_utils import PasswordUtils
from archipy.models.errors import InvalidPasswordError

# Configure logging
logger = logging.getLogger(__name__)

# Hash a password
password = "SecureP@ssword123"
hashed = PasswordUtils.hash_password(password)

# Verify password
is_valid = PasswordUtils.verify_password(password, hashed)
logger.info(f"Password valid: {is_valid}")

# Generate a secure password that meets policy requirements
secure_password = PasswordUtils.generate_password()
logger.info(f"Generated password: {secure_password}")

# Validate a password against policy
try:
    PasswordUtils.validate_password(password)
except InvalidPasswordError as e:
    requirements = e.additional_data.get("requirements", [])
    logger.warning(f"Invalid password. Requirements not met: {requirements}")
    raise
else:
    logger.info("Password meets policy requirements")

# Check password against history
password_history = [hashed]  # Previous password hashes
try:
    PasswordUtils.validate_password_history("NewSecureP@ssword123", password_history)
except InvalidPasswordError as e:
    logger.warning("Password has been used recently")
    raise
else:
    logger.info("Password not previously used")

File Utils¶

Handle files securely:

import logging

from archipy.helpers.utils.file_utils import FileUtils
from archipy.models.errors import InvalidArgumentError, OutOfRangeError

# Configure logging
logger = logging.getLogger(__name__)

# Create a secure link to a file with expiration
try:
    link = FileUtils.create_secure_link("/path/to/document.pdf", minutes=60)
except (InvalidArgumentError, OutOfRangeError) as e:
    logger.error(f"Error creating link: {e}")
    raise
else:
    logger.info(f"Secure link: {link}")

# Validate file name against allowed extensions
try:
    is_valid = FileUtils.validate_file_name("document.pdf")
except InvalidArgumentError as e:
    logger.error(f"Error validating file: {e}")
    raise
else:
    logger.info(f"File is valid: {is_valid}")

Base Utils¶

Validate and sanitize data:

import logging

from archipy.helpers.utils.base_utils import BaseUtils
from archipy.models.errors import InvalidArgumentError

# Configure logging
logger = logging.getLogger(__name__)

# Sanitize phone number
phone = BaseUtils.sanitize_iranian_landline_or_phone_number("+989123456789")
logger.info(f"Sanitized phone: {phone}")  # 09123456789

# Validate Iranian national code
try:
    BaseUtils.validate_iranian_national_code_pattern("1234567891")
except InvalidArgumentError as e:
    logger.error(f"Invalid national code: {e}")
    raise
else:
    logger.info("National code is valid")

Error Utils¶

Utilities for standardized error introspection. See the Error Handling for comprehensive patterns using ArchiPy's domain-specific exceptions.

App Utils¶

FastAPI application utilities:

import logging

from archipy.helpers.utils.app_utils import AppUtils, FastAPIUtils
from archipy.configs.base_config import BaseConfig

# Configure logging
logger = logging.getLogger(__name__)

# Create a FastAPI app with standard config
app = AppUtils.create_fastapi_app(BaseConfig.global_config())

# AppUtils.create_fastapi_app already registers the exception handlers
# (archipy.helpers.errors.fastapi.setup_exception_handlers); pass configure_exception_handlers=False to skip them.


# Set up CORS (reads origins from config.FASTAPI.CORS_MIDDLEWARE_ALLOW_ORIGINS)
FastAPIUtils.setup_cors(app, BaseConfig.global_config())

logger.info("FastAPI app configured successfully")

String Utils¶

String manipulation utilities. StringUtils provides helpers for common string operations. See the Utils API Reference for the full method list.

Keycloak Utils¶

Authentication and authorization utilities with Keycloak integration:

Moved to the Keycloak adapter

The guards are now the functions fastapi_auth and grpc_auth of archipy.adapters.keycloak.auth. fastapi_auth returns an async dependency and grpc_auth detects async def servicer methods, so async_fastapi_auth, async_grpc_auth and the KeycloakUtils class (in archipy.helpers.utils.keycloak_utils) are deprecated and emit a DeprecationWarning. They will be removed in 6.0.0, together with the archipy.helpers.utils.keycloak_utils module.

import logging
from uuid import UUID

import uvicorn
from fastapi import Depends

from archipy.configs.base_config import BaseConfig
from archipy.helpers.utils.app_utils import AppUtils
from archipy.adapters.keycloak.auth import fastapi_auth
from archipy.models.types.language_type import LanguageType

# Configure logging
logger = logging.getLogger(__name__)

# Initialize your app configuration
config = BaseConfig()
BaseConfig.set_global(config)
app = AppUtils.create_fastapi_app()

# Resource-based authorization for users with role and admin access
@app.get("/users/{user_uuid}/info")
def get_user_info(user_uuid: UUID, user: dict = Depends(fastapi_auth(
    resource_type_param="user_uuid",
    resource_type="users",
    required_roles={"user"},
    admin_roles={"superusers", "administrators"},
    lang=LanguageType.EN,
))):
    return {
        "message": f"User info for {user_uuid}",
        "username": user.get("preferred_username")
    }

# The dependency is async, so it works the same for def and async def endpoints
@app.get("/employees/{employee_uuid}/info")
async def get_employee_info(employee_uuid: UUID, employee: dict = Depends(fastapi_auth(
    resource_type_param="employee_uuid",
    resource_type="employees",
    required_roles={"employee", "manager", "user"},
    all_roles_required=False,  # User can have any of these roles
    admin_roles={"hr_admins", "system_admins"},
    lang=LanguageType.FA,
))):
    return {
        "message": f"Employee info for {employee_uuid}",
        "username": employee.get("preferred_username")
    }

uvicorn.run(app, host="0.0.0.0", port=8000)  # noqa: S104

fastapi_auth returns the UserInfo dict, and it sets AuthContextManager for the request, so code below the endpoint can read the caller without receiving the request:

from archipy.adapters.keycloak.auth import AuthContextManager

def audit_log(action: str) -> None:
    caller = AuthContextManager.get_auth_context()  # AuthenticatedUserDTO, or None outside a guarded request
    logger.info("%s by %s", action, caller.username if caller else "anonymous")

The context is visible to async def and def endpoints, background tasks and streamed responses, and it is cleared when the request ends.

request.state is deprecated for the auth data

The dependency still fills request.state.user_info, token_info, actor_chain and user_roles, but each of them is now a dict, list or set that emits a DeprecationWarning when you read it (iterate, index, get, in, len, ...), naming its replacement on AuthContextManager.get_auth_context(): raw_user_info, token_info, actor_chain and roles. The values are the same as before. The dict the dependency returns does not warn. request.state will stop carrying this data in 6.0.0.