from __future__ import annotations import datetime, pytz, redis from passlib.hash import argon2 #from datetime import datetime, time, timedelta from fastapi import APIRouter, Depends, Header, HTTPException, status from pydantic import BaseModel, EmailStr, Field from typing import Dict, List, Optional, Set, Union from .log import * from .db_sql import sql_select # ### BEGIN ### API Lib General ### async get_token_header() ### async def get_token_header(x_token:str = Header(...)): if x_token != 'fake-super-secret-token': raise HTTPException(status_code=400, detail='X-Token header invalid') # ### END ### API Lib General ### async get_token_header() ### # ### BEGIN ### API Lib General ### async get_account_header() ### async def get_account_header(x_account_id:str = Header(...)): log.setLevel(logging.WARNING) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) if len(x_account_id): log.info('The x-account-id header has a value.') if account_id := redis_lookup_id_random(table_name='account', record_id_random=x_account_id): log.setLevel(logging.DEBUG) log.info('Found the account_id with the account_id_random value: '+x_account_id) account = { 'id': account_id, 'id_random': x_account_id } else: log.setLevel(logging.DEBUG) log.info('The x-account-id was invalid and not empty...') #raise HTTPException(status_code=500) raise HTTPException(status_code=400) # or 404? #return False elif x_account_id == '': log.info('The x-account-id header was empty.') account = { 'id': None, 'id_random': None } return account # ### END ### API Lib General ### async get_account_header() ### def secure_hash_string(string:str): string_hash = argon2.using(rounds=14, memory_cost=1536, parallelism=2).hash(string) return string_hash def verify_secure_hash_string(string:str, string_hash:str): if argon2.verify(string, string_hash): return True else: return False