utils

utils

Utilities Library

A comprehensive collection of utility functions for data processing, file operations, image handling, and common programming tasks. This library is designed to be standalone and reusable across different projects.

Modules:

Name	Description
chunk_execution	Async execution utilities with retry logic and concurrency control
compare	Data comparison utilities for dictionaries and lists
convert	Data conversion utilities for various formats and types
DictDot	Dot notation access for dictionaries
files	File and folder operation utilities
images	Image processing and manipulation utilities
logging	Custom logging processors and utilities for crew_dcs
read_creds_from_dotenv	Environment credential reading utilities
upload_data	Data upload utilities (may require external dependencies)
xkcd_password	Password generation utilities
exceptions	Custom exception classes for error handling

Usage

from utils import chunk_execution, convert, DictDot from utils.exceptions import UtilityError

Use async utilities

await chunk_execution.run_with_retry(my_function)

Convert data formats

datetime_obj = convert.convert_string_to_datetime("2023-01-01")

Use dot notation for dictionaries

data = DictDot.DictDot({"user": {"name": "John"}}) print(data.user.name) # "John"

Version: 1.0.0 Author: Domo Utils Contributors License: MIT

ConcatDataframeError

ConcatDataframeError(
    element: Any, operation: str = "concatenation"
)

Bases: UtilityError

Raised when dataframe concatenation operations fail.

This exception is raised when attempting to concatenate objects that are not pandas DataFrames or when concatenation operations fail.

Parameters:

Name	Type	Description	Default
element	Any	The invalid element that caused the error	required
operation	str	The operation being performed	'concatenation'

Example

try: ... concat_dataframes([df1, "not_a_dataframe", df2]) ... except ConcatDataframeError as e: ... print(f"Error: {e}") Error: Invalid element type for concatenation:

Source code in src/crew_dcs/utils/exceptions.py

def __init__(self, element: Any, operation: str = "concatenation"):
    element_type = type(element).__name__
    message = f"Invalid element type for {operation}: {element_type}"
    super().__init__(message, {"element": element, "type": element_type})
    self.element = element
    self.operation = operation

CredentialsError

CredentialsError(source: str, details: Any | None = None)

Bases: UtilityError

Raised when credential reading or validation fails.

This exception is raised when attempting to read credentials from environment files or when credential validation fails.

Parameters:

Name	Type	Description	Default
source	str	The credential source (file path, environment, etc.)	required
details	Any	Additional error details	None

Example

try: ... read_credentials(".env") ... except CredentialsError as e: ... print(f"Error: {e}") Error: Failed to read credentials from ".env"

Source code in src/crew_dcs/utils/exceptions.py

def __init__(self, source: str, details: Any | None = None):
    message = f'Failed to read credentials from "{source}"'
    super().__init__(message, details)
    self.source = source

FileOperationError

FileOperationError(
    operation: str,
    file_path: str,
    details: Any | None = None,
)

Bases: UtilityError

Raised when file operations fail.

This exception is raised when file creation, reading, writing, or other file system operations encounter errors.

Parameters:

Name	Type	Description	Default
operation	str	The file operation that failed	required
file_path	str	The file path involved in the operation	required
details	Any	Additional error details	None

Example

try: ... create_folder("/invalid/path") ... except FileOperationError as e: ... print(f"Error: {e}") Error: Failed to create folder at "/invalid/path"

Source code in src/crew_dcs/utils/exceptions.py

def __init__(self, operation: str, file_path: str, details: Any | None = None):
    message = f'Failed to {operation} at "{file_path}"'
    super().__init__(message, details)
    self.operation = operation
    self.file_path = file_path

ImageProcessingError

ImageProcessingError(
    operation: str, details: Any | None = None
)

Bases: UtilityError

Raised when image processing operations fail.

This exception is raised when image loading, processing, or manipulation operations encounter errors.

Parameters:

Name	Type	Description	Default
operation	str	The image operation that failed	required
details	Any	Additional error details	None

Example

try: ... load_image("invalid.jpg") ... except ImageProcessingError as e: ... print(f"Error: {e}") Error: Failed to load image

Source code in src/crew_dcs/utils/exceptions.py

def __init__(self, operation: str, details: Any | None = None):
    message = f"Failed to {operation}"
    super().__init__(message, details)
    self.operation = operation

InvalidEmailError

InvalidEmailError(email: str)

Bases: UtilityError

Raised when email validation fails.

This exception is raised when a provided email address does not match the expected email format pattern.

Parameters:

Name	Type	Description	Default
email	str	The invalid email address that caused the error	required

Example

try: ... validate_email("invalid-email") ... except InvalidEmailError as e: ... print(f"Error: {e}") Error: Invalid email format: "invalid-email"

Source code in src/crew_dcs/utils/exceptions.py

def __init__(self, email: str):
    message = f'Invalid email format: "{email}"'
    super().__init__(message, {"email": email})
    self.email = email

UtilityError

UtilityError(message: str, details: Any | None = None)

Bases: Exception

Base exception class for all utility errors.

Provides consistent error handling and message formatting across all utilities. All other utility exceptions should inherit from this class.

Parameters:

Name	Type	Description	Default
message	str	The error message	required
details	Any	Additional error details for debugging	None

Attributes:

Name	Type	Description
message	str	The error message
details	Any	Additional error details

Source code in src/crew_dcs/utils/exceptions.py

def __init__(self, message: str, details: Any | None = None):
    self.message = message
    self.details = details
    super().__init__(self.message)

Modules

• DictDot
• chunk_execution
• compare
• convert
• enums
• exceptions
• federation_utils
• files
• images
• logging
• quartz_cron
• read_creds_from_dotenv
• sample_data
• upload_data
• xkcd_password