Skip to content

📝 Add docstrings to feature/auth-service-workflow #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
coderabbitai wants to merge 1 commit into from
Closed

📝 Add docstrings to feature/auth-service-workflow #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 23 additions & 27 deletions backend/app/auth/routes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,16 +15,13 @@
@router.post("/signup/email", response_model=AuthResponse)
async def signup_with_email(request: EmailSignupRequest):
"""
Registers a new user using email, password, and name, and returns authentication tokens and user information.
Register a new user with email, password, and name, returning authentication tokens and user information.

Args:
request: Contains the user's email, password, and name for registration.
Parameters:
request (EmailSignupRequest): Registration details including email, password, and name.

Returns:
An AuthResponse with access token, refresh token, and user details.

Raises:
HTTPException: If registration fails or an unexpected error occurs.
AuthResponse: Contains access token, refresh token, and user details.
"""
try:
result = await auth_service.create_user_with_email(
Expand Down Expand Up @@ -58,9 +55,10 @@ async def signup_with_email(request: EmailSignupRequest):
@router.post("/login/email", response_model=AuthResponse)
async def login_with_email(request: EmailLoginRequest):
"""
Authenticates a user using email and password credentials.
Authenticate a user with email and password, returning access and refresh tokens along with user details.

On successful authentication, returns an access token, refresh token, and user information. Raises an HTTP 500 error if authentication fails due to an unexpected error.
Returns:
AuthResponse: Contains the JWT access token, refresh token, and authenticated user information.
"""
try:
result = await auth_service.authenticate_user_with_email(
Expand Down Expand Up @@ -93,9 +91,10 @@ async def login_with_email(request: EmailLoginRequest):
@router.post("/login/google", response_model=AuthResponse)
async def login_with_google(request: GoogleLoginRequest):
"""
Authenticates or registers a user using a Google OAuth ID token.
Authenticate or register a user using a Google OAuth ID token.

On success, returns an access token, refresh token, and user information. Raises an HTTP 500 error if Google authentication fails.
Returns:
AuthResponse: Contains a JWT access token, refresh token, and user information upon successful authentication or registration.
"""
try:
result = await auth_service.authenticate_with_google(request.id_token)
Expand Down Expand Up @@ -125,12 +124,12 @@ async def login_with_google(request: GoogleLoginRequest):
@router.post("/refresh", response_model=TokenResponse)
async def refresh_token(request: RefreshTokenRequest):
"""
Refreshes JWT tokens using a valid refresh token.
Refreshes JWT access and refresh tokens using a valid refresh token.

Validates the provided refresh token, issues new tokens if valid, and returns them. Raises a 401 error if the refresh token is invalid or revoked.

Validates the provided refresh token, issues a new access token and refresh token if valid, and returns them. Raises a 401 error if the refresh token is invalid or revoked.

Returns:
A TokenResponse containing the new access and refresh tokens.
TokenResponse: Contains the new access and refresh tokens.
"""
try:
new_refresh_token = await auth_service.refresh_access_token(request.refresh_token)
Expand Down Expand Up @@ -169,10 +168,13 @@ async def refresh_token(request: RefreshTokenRequest):
@router.post("/token/verify", response_model=UserResponse)
async def verify_token(request: TokenVerifyRequest):
"""
Verifies an access token and returns the associated user information.
Verify an access token and return the corresponding user information.

Returns:
UserResponse: User data associated with the valid access token.

Raises:
HTTPException: If the token is invalid or expired, returns a 401 Unauthorized error.
HTTPException: Returns 401 Unauthorized if the token is invalid or expired.
"""
try:
user = await auth_service.verify_access_token(request.access_token)
Expand All @@ -192,10 +194,10 @@ async def verify_token(request: TokenVerifyRequest):
@router.post("/password/reset/request", response_model=SuccessResponse)
async def request_password_reset(request: PasswordResetRequest):
"""
Initiates a password reset process by sending a reset link to the provided email address.
Initiate a password reset by sending a reset link to the specified email address.

Returns:
SuccessResponse: Indicates whether the password reset email was sent if the email exists.
SuccessResponse: Indicates that a reset link has been sent if the email exists in the system.
"""
try:
await auth_service.request_password_reset(request.email)
Expand All @@ -212,16 +214,10 @@ async def request_password_reset(request: PasswordResetRequest):
@router.post("/password/reset/confirm", response_model=SuccessResponse)
async def confirm_password_reset(request: PasswordResetConfirm):
"""
Resets a user's password using a valid password reset token.

Args:
request: Contains the password reset token and the new password.
Reset a user's password using a valid reset token and new password.

Returns:
SuccessResponse indicating the password has been reset successfully.

Raises:
HTTPException: If the reset token is invalid or an error occurs during the reset process.
SuccessResponse: Indicates the password has been reset successfully.
"""
try:
await auth_service.confirm_password_reset(
Expand Down
43 changes: 20 additions & 23 deletions backend/app/auth/security.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,41 +15,39 @@

def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Verifies whether a plaintext password matches a given hashed password.
Check if a plaintext password matches a given bcrypt hashed password.

Args:
plain_password: The plaintext password to verify.
hashed_password: The hashed password to compare against.
Parameters:
plain_password (str): The plaintext password to verify.
hashed_password (str): The bcrypt hashed password to compare against.

Returns:
True if the plaintext password matches the hash, otherwise False.
bool: True if the plaintext password matches the hashed password, otherwise False.
"""
return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
"""
Hashes a plaintext password using bcrypt.
Hash a plaintext password using bcrypt and return the hashed string.

Args:
password: The plaintext password to hash.
Parameters:
password (str): The plaintext password to be hashed.

Returns:
The bcrypt-hashed password as a string.
str: The bcrypt hash of the provided password.
"""
return pwd_context.hash(password)

def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
"""
Creates a JWT access token embedding the provided data and an expiration time.
Generate a signed JWT access token containing the provided payload and an expiration time.

If `expires_delta` is not specified, the token expires after the default duration from settings. The payload includes an expiration timestamp and a type field set to "access". The token is signed using the configured secret key and algorithm.

Args:
data: The payload to include in the token.
expires_delta: Optional timedelta specifying how long the token is valid.
Parameters:
data (Dict[str, Any]): The payload to embed in the token.
expires_delta (Optional[timedelta]): Optional duration for which the token remains valid. If not provided, a default expiration from settings is used.

Returns:
A signed JWT access token as a string.
str: The encoded JWT access token as a string.
"""
to_encode = data.copy()
if expires_delta:
Expand All @@ -63,19 +61,18 @@ def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta]

def create_refresh_token() -> str:
"""
Generates a secure random refresh token as a URL-safe string.
Generate a cryptographically secure, URL-safe random string for use as a refresh token.

Returns:
A cryptographically secure, URL-safe refresh token string.
str: A secure, URL-safe refresh token string.
"""
return secrets.token_urlsafe(32)

def verify_token(token: str) -> Dict[str, Any]:
"""
Verifies and decodes a JWT token.
Decode and validate a JWT token, returning its payload as a dictionary.

If the token is invalid or cannot be verified, raises an HTTP 401 Unauthorized exception.
Returns the decoded token payload as a dictionary.
Raises an HTTP 401 Unauthorized exception if the token is invalid or verification fails.
"""
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
Expand All @@ -89,9 +86,9 @@ def verify_token(token: str) -> Dict[str, Any]:

def generate_reset_token() -> str:
"""
Generates a secure, URL-safe token for password reset operations.
Generate a cryptographically secure, URL-safe random string for use as a password reset token.

Returns:
A random 32-byte URL-safe string suitable for use as a password reset token.
str: A random 32-byte URL-safe string suitable for password reset operations.
"""
return secrets.token_urlsafe(32)
80 changes: 40 additions & 40 deletions backend/app/auth/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,30 +53,30 @@
class AuthService:
def __init__(self):
# Initializes the AuthService instance.
"""
Initialize a new instance of the AuthService class.
"""
pass

def get_db(self):
"""
Returns a database connection instance from the application's database module.
Retrieve a database connection instance from the application's database module.
"""
return get_database()

async def create_user_with_email(self, email: str, password: str, name: str) -> Dict[str, Any]:
"""
Creates a new user account with the provided email, password, and name.
Create a new user account with the specified email, password, and name.

Checks for existing users with the same email and raises an error if found. Stores the user with a hashed password and default profile fields, then generates and returns a refresh token along with the user data.
Checks for an existing user with the given email and raises an HTTP 400 error if found. Stores the user with a hashed password and default profile fields, then generates and returns a refresh token along with the user data.

Args:
email: The user's email address.
password: The user's plaintext password.
name: The user's display name.
Parameters:
email (str): The user's email address.
password (str): The user's plaintext password.
name (str): The user's display name.

Returns:
A dictionary containing the created user document and a refresh token.

Raises:
HTTPException: If a user with the given email already exists.
Dict[str, Any]: A dictionary containing the created user document and a refresh token.
"""
db = self.get_db()

Expand Down Expand Up @@ -119,12 +119,12 @@ async def create_user_with_email(self, email: str, password: str, name: str) ->

async def authenticate_user_with_email(self, email: str, password: str) -> Dict[str, Any]:
"""
Authenticates a user using email and password credentials.
Authenticate a user by verifying email and password credentials.

Verifies the provided email and password against stored user data. If authentication succeeds, returns the user information and a new refresh token. Raises an HTTP 401 error if credentials are invalid.
If authentication is successful, returns a dictionary containing the user document and a new refresh token. Raises an HTTP 401 error if the credentials are invalid.

Returns:
A dictionary containing the authenticated user and a new refresh token.
dict: Contains the authenticated user and a new refresh token.
"""
db = self.get_db()

Expand All @@ -145,15 +145,15 @@ async def authenticate_user_with_email(self, email: str, password: str) -> Dict[

async def authenticate_with_google(self, id_token: str) -> Dict[str, Any]:
"""
Authenticates a user using a Google OAuth ID token, creating a new user if necessary.
Authenticate a user using a Google OAuth ID token, creating or updating the user as needed.

Verifies the provided Firebase ID token, retrieves or creates the corresponding user in the database, updates user information if needed, and issues a new refresh token. Raises an HTTP 400 error if the email is missing or if authentication fails, and HTTP 401 if the token is invalid.
Verifies the provided Firebase ID token, retrieves or creates the corresponding user in the database, updates user information if necessary, and issues a new refresh token. Raises an HTTP 400 error if the email is missing or authentication fails, and HTTP 401 if the token is invalid.

Args:
id_token: The Firebase ID token obtained from Google OAuth.
Parameters:
id_token (str): The Firebase ID token obtained from Google OAuth.

Returns:
A dictionary containing the user data and a new refresh token.
Dict[str, Any]: A dictionary containing the user data and a new refresh token.
"""
try:
# Verify the Firebase ID token
Expand Down Expand Up @@ -229,15 +229,15 @@ async def authenticate_with_google(self, id_token: str) -> Dict[str, Any]:

async def refresh_access_token(self, refresh_token: str) -> str:
"""
Refreshes an access token by validating and rotating the provided refresh token.
Refreshes the access token by validating and rotating the provided refresh token.

If the refresh token is valid and not expired, issues a new refresh token and revokes the old one. Raises an HTTP 401 error if the token is invalid, expired, or the associated user does not exist.

Args:
refresh_token: The refresh token string to validate and rotate.
Parameters:
refresh_token (str): The refresh token to validate and rotate.

Returns:
A new refresh token string.
str: A new refresh token string.
"""
db = self.get_db()

Expand Down Expand Up @@ -274,13 +274,13 @@ async def refresh_access_token(self, refresh_token: str) -> str:
return new_refresh_token
async def verify_access_token(self, token: str) -> Dict[str, Any]:
"""
Verifies an access token and retrieves the associated user.
Verify a JWT access token and return the associated user document.

Args:
token: The JWT access token to verify.
Parameters:
token (str): The JWT access token to verify.

Returns:
The user document corresponding to the token's subject.
Dict[str, Any]: The user document corresponding to the token's subject.

Raises:
HTTPException: If the token is invalid or the user does not exist.
Expand Down Expand Up @@ -309,9 +309,9 @@ async def verify_access_token(self, token: str) -> Dict[str, Any]:

async def request_password_reset(self, email: str) -> bool:
"""
Initiates a password reset process for the specified email address.
Initiates a password reset process for the given email address.

If the user exists, generates a password reset token with a 1-hour expiration and stores it in the database. The reset token and link are logged for development purposes. Always returns True to avoid revealing whether the email is registered.
If the user exists, generates and stores a password reset token with a 1-hour expiration. Always returns True to prevent email enumeration.
"""
db = self.get_db()

Expand Down Expand Up @@ -342,16 +342,16 @@ async def request_password_reset(self, email: str) -> bool:

async def confirm_password_reset(self, reset_token: str, new_password: str) -> bool:
"""
Confirms a password reset using a valid reset token and updates the user's password.
Reset a user's password using a valid reset token and revoke all existing refresh tokens.

Validates the reset token, updates the user's password, marks the token as used, and revokes all existing refresh tokens for the user to require re-authentication.
Validates the provided reset token, updates the user's password, marks the token as used, and revokes all refresh tokens for the user to require re-authentication.

Args:
reset_token: The password reset token to validate.
new_password: The new password to set for the user.
Parameters:
reset_token (str): The password reset token to validate and consume.
new_password (str): The new password to set for the user.

Returns:
True if the password reset is successful.
bool: True if the password reset is successful.

Raises:
HTTPException: If the reset token is invalid or expired.
Expand Down Expand Up @@ -393,15 +393,15 @@ async def confirm_password_reset(self, reset_token: str, new_password: str) -> b
return True
async def _create_refresh_token_record(self, user_id: str) -> str:
"""
Generates and stores a new refresh token for the specified user.
Generate and store a new refresh token for a user, returning the token string.

Creates a refresh token with an expiration date and saves it in the database for token management and rotation.
A refresh token with an expiration date is created and saved in the database for the specified user. The token is used for session management and token rotation.

Args:
user_id: The unique identifier of the user for whom the refresh token is created.
Parameters:
user_id (str): Unique identifier of the user for whom the refresh token is generated.

Returns:
The generated refresh token string.
str: The generated refresh token.
"""
db = self.get_db()

Expand Down
Loading