services.auth#

Module Contents#

Classes#

Auth

Class handling authentication operations such as password hashing, JWT token creation, and token blacklisting.

Data#

auth_service

API#

class services.auth.Auth#

Class handling authentication operations such as password hashing, JWT token creation, and token blacklisting.

pwd_context = 'CryptContext(...)'#
oauth2_scheme = 'OAuth2PasswordBearer(...)'#
SECRET_KEY = None#
ALGORITHM = None#
verify_password(plain_password: str, hashed_password: str)#

Verify the given plain password against the hashed password.

Parameters:

  • plain_password (str) – Plain text password.

  • hashed_password (str) – Hashed password.

Returns:

True if the passwords match, False otherwise.

Return type:

bool

get_password_hash(password: str)#

Generate the hash for the given password.

Parameters:

password (str) – Plain text password.

Returns:

Hashed password.

Return type:

str

async create_access_token(data: dict, expires_delta: Optional[float] = None)#

Create an access token.

Parameters:

  • data (dict) – Payload data to be encoded in the token.

  • expires_delta (Optional[float]) – Expiry time for the token.

Returns:

Encoded access token.

Return type:

str

async create_refresh_token(data: dict, expires_delta: Optional[float] = None)#

Create a refresh token.

Parameters:

  • data (dict) – Payload data to be encoded in the token.

  • expires_delta (Optional[float]) – Expiry time for the token.

Returns:

Encoded refresh token.

Return type:

str

async decode_refresh_token(refresh_token: str)#

Decode the refresh token and retrieve the email from the payload.

Parameters:

refresh_token (str) – Encoded refresh token.

Returns:

Email extracted from the token payload.

Return type:

str

async static add_token_to_blacklist(user_id: int, token: str, db: sqlalchemy.ext.asyncio.AsyncSession = Depends(get_db))#

Add a token to the blacklist.

Parameters:

  • user_id (int) – User ID associated with the token.

  • token (str) – Token to be blacklisted.

  • db – Async database session.

async static is_token_blacklisted(token: str, db: sqlalchemy.ext.asyncio.AsyncSession = Depends(get_db))#

Check if a token is blacklisted.

Parameters:

  • token (str) – Token to be checked.

  • db (AsyncSession) – Async database session.

Returns:

Blacklisted token record if found, None otherwise.

Return type:

Blacklisted | None

async get_current_user(token: str = Depends(oauth2_scheme), db: sqlalchemy.ext.asyncio.AsyncSession = Depends(get_db))#

Get the current authenticated user.

Parameters:

  • token (str) – Encoded JWT token.

  • db (AsyncSession) – Async database session.

Returns:

Current authenticated user.

Return type:

User

services.auth.auth_service = 'Auth(...)'#