Config

store_credential(service, username, value, field='password', backend='1password', vault=None, account=None)[source]

Store credential in specified backend.

Parameters:

service (str) – Service name
username (str) – Username or identifier
value (str) – Credential value
field (str) – Field name
backend (str) – Backend to use (‘1password’, ‘keychain’, ‘env’)
vault (str | None) – 1Password vault override (falls back to default_vault)
account (str | None) – 1Password account override (falls back to default_account)

Returns:

True if successful, False otherwise

Return type:

store_google_analytics_credentials(credentials_data, item_title='Google Analytics API', vault=None)[source]

Store Google Analytics OAuth credentials.

Parameters:

credentials_data (Dict[str, Any]) – OAuth2 credentials from Google Cloud Console
item_title (str) – Title for the credential item
vault (str | None) – 1Password vault (uses default if not specified)

Returns:

True if successful, False otherwise

Return type:

get_google_analytics_credentials(item_title='Google Analytics API')[source]

Get Google Analytics credentials for GoogleAnalyticsConnector.

Parameters:: item_title (str) – Title of the credential item
Returns:: Tuple of (client_id, client_secret).
Raises:: CredentialNotFoundError – when neither 1Password item lookup nor the general credential backends yield both client_id and client_secret. The error’s field attribute names which field was missing; attempts carries per-backend diagnostics so callers can surface an actionable message.
Return type:: Tuple[str, str]

list_stored_credentials(service_filter=None, vault=None, account=None)[source]

List stored credentials across all backends.

Parameters:

service_filter (str | None) – Optional service name to filter by
vault (str | None) – 1Password vault override (falls back to default_vault)
account (str | None) – 1Password account override (falls back to default_account)

Returns:

List of credential information dictionaries

Return type:

List[Dict[str, Any]]

backend_status()[source]

Get detailed status of all credential backends.

Return type:: Dict[str, Dict[str, Any]]

siege_utilities.config.credential_manager.get_credential(service, username, field='password', search_paths=None, vault=None, account=None)[source]

Get credential using default credential manager.

Parameters:

service (str) – Service name (or 1Password item title)
username (str) – Username or identifier
field (str) – Field to retrieve
search_paths (List[str | Path] | None) – Additional paths to search for credential files
vault (str | None) – 1Password vault (overrides default “Private”)
account (str | None) – 1Password account shorthand or UUID

Returns:

Credential value as a string.

Raises:

CredentialNotFoundError – when no configured backend yields the credential. See CredentialManager.get_credential for the attempt-log shape.
Exception – backend-specific transport / auth errors propagate from the underlying helpers (e.g., 1Password CLI nonzero exits other than “not found”).

Return type:

siege_utilities.config.credential_manager.store_credential(service, username, value, field='password', backend='1password', vault=None, account=None)[source]

Store credential using default credential manager.

Parameters:

service (str) – Service name
username (str) – Username or identifier
value (str) – Credential value
field (str) – Field name
backend (str) – Backend to use
vault (str | None) – 1Password vault (overrides default “Private”)
account (str | None) – 1Password account shorthand or UUID

Returns:

True if successful

Return type:

siege_utilities.config.credential_manager.store_ga_credentials_from_file(credentials_file, item_title='Google Analytics API - Multi-Client Reporter', vault='Private', delete_file=True)[source]

Store Google Analytics credentials from JSON file.

Parameters:

credentials_file (str | Path) – Path to OAuth2 JSON file
item_title (str) – Title for 1Password item
vault (str) – 1Password vault
delete_file (bool) – Whether to delete source file

Returns:

True if successful

Return type:

siege_utilities.config.credential_manager.get_ga_credentials()[source]

Get Google Analytics credentials for connector.

Returns:: Tuple of (client_id, client_secret).
Raises:: CredentialNotFoundError – when no configured backend yields both client_id and client_secret. See CredentialManager.get_google_analytics_credentials for details.
Return type:: Tuple[str, str]

siege_utilities.config.credential_manager.credential_status()[source]

Get status of all credential backends.

Return type:: Dict[str, Dict[str, Any]]

siege_utilities.config.credential_manager.store_ga_service_account_from_file(credentials_file, item_title='Google Analytics Service Account', vault='Private', delete_file=False)[source]

Store Google Analytics service account credentials in 1Password.

Parameters:

credentials_file (str | Path) – Path to service account JSON file
item_title (str) – Title for 1Password item
vault (str) – 1Password vault name
delete_file (bool) – Whether to delete file after storing

Returns:

True if successful

Return type:

siege_utilities.config.credential_manager.get_ga_service_account_credentials()[source]

Get Google Analytics service account credentials.

Returns:: Dict with keys client_email, project_id, private_key, private_key_id, type (always "service_account").
Raises:: CredentialNotFoundError – when any required field cannot be retrieved from any configured backend. The error’s field attribute names which field was missing; attempts carries per-backend diagnostics so callers can surface an actionable message instead of a generic “service account not configured.”
Return type:: Dict[str, str]

siege_utilities.config.credential_manager.get_google_service_account_from_1password(item_title='Google Analytics Service Account - Multi-Client Reporter', vault=None, account=None)[source]

Get Google service account credentials from 1Password. Based on working implementation from GA project.

Parameters:

item_title (str) – Title of the 1Password item containing service account
vault (str | None) – 1Password vault name
account (str | None) – 1Password account shorthand or UUID

Returns:

Service account credentials dictionary.

Raises:

subprocess.CalledProcessError – If 1Password CLI fails (item not found, auth required, etc.)
subprocess.TimeoutExpired – If the op command exceeds 60 s.
OSError – If the op binary cannot be executed.

Return type:

siege_utilities.config.credential_manager.get_google_oauth_from_1password(item_title='Google Analytics API - Multi-Client Reporter', vault=None, account=None)[source]

Get Google OAuth2 client credentials from 1Password.

Returns dict with client_id, client_secret, project_id, redirect_uri suitable for GoogleAnalyticsConnector(auth_method=”oauth2”).

Parameters:

item_title (str) – Title of the 1Password item containing OAuth2 credentials
vault (str | None) – 1Password vault name (e.g., ‘Private’)
account (str | None) – 1Password account shorthand or UUID

Returns:

Dict with OAuth2 credential fields.

Raises:

subprocess.CalledProcessError – If 1Password CLI fails.
subprocess.TimeoutExpired – If the op command exceeds 60 s.
ValueError – If the item exists but lacks client_id or client_secret.
OSError – If the op binary cannot be executed.

Return type:

siege_utilities.config.credential_manager.get_google_oauth_document_from_1password(item_title='Google OAuth Client - siege_utilities', vault=None, account=None)[source]

Get Google OAuth2 client secret JSON from a 1Password DOCUMENT item.

Unlike get_google_oauth_from_1password() which reads individual fields, this function downloads an attached JSON file (client_secret_*.json) stored as a 1Password Document item.

Parameters:

item_title (str) – Title of the 1Password Document item.
vault (str | None) – 1Password vault name.
account (str | None) – 1Password account shorthand or UUID.

Returns:

Parsed client secret dict (contains "installed" or "web" key).

Raises:

subprocess.CalledProcessError – If 1Password CLI fails (item not found, auth required, etc.)
subprocess.TimeoutExpired – If the op command exceeds 60 s.
json.JSONDecodeError – If the document content is not valid JSON.
OSError – If the op binary cannot be executed.

Return type:

siege_utilities.config.credential_manager.create_temporary_service_account_file(service_account_data)[source]

Create a temporary service account file for Google APIs. Useful for APIs that require a file path.

Parameters:

service_account_data (Dict[str, str]) – Service account credentials dictionary

Returns:

Path to temporary file.

Raises:

OSError – If the temporary file cannot be created.
TypeError – If service_account_data is not JSON-serializable.

Return type:

Connection management and persistence for siege_utilities. Handles notebook connections, Spark connections, and their persistence.

siege_utilities.config.connections.create_connection_profile(name, connection_type, connection_params, **kwargs)[source]

Create a connection profile for various connection types.

Parameters:

name (str) – Friendly name for the connection
connection_type (str) – Type of connection (notebook, spark, database, api, etc.)
connection_params (Dict[str, Any]) – Connection-specific parameters
**kwargs – Additional connection settings

Returns:

Dictionary with connection profile configuration

Return type:

Example

>>> import siege_utilities
>>> notebook_conn = siege_utilities.create_connection_profile(
...     "Jupyter Lab",
...     "notebook",
...     {
...         "url": "http://localhost:8888",
...         "token": "abc123",
...         "workspace": "/home/user/notebooks"
...     },
...     auto_connect=True
... )

siege_utilities.config.connections.save_connection_profile(profile, config_directory='config')[source]

Save connection profile to JSON file.

Parameters:

profile (Dict[str, Any]) – Connection profile dictionary
config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Return type:

Example

>>> conn = create_connection_profile("Test", "notebook", {"url": "http://test"})
>>> file_path = siege_utilities.save_connection_profile(conn)

siege_utilities.config.connections.load_connection_profile(connection_id, config_directory='config')[source]

Load connection profile from JSON file.

Parameters:

connection_id (str) – Connection ID to load
config_directory (str) – Directory containing config files

Returns:

Connection profile dictionary.

Raises:

FileNotFoundError – If the connection profile does not exist.
json.JSONDecodeError – If the file contains invalid JSON.
OSError – If the file cannot be read.

Return type:

Example

>>> profile = siege_utilities.load_connection_profile("uuid-here")
>>> print(f"Loaded: {profile['name']}")

siege_utilities.config.connections.find_connection_by_name(name, config_directory='config')[source]

Find a connection profile by name.

Parameters:

name (str) – Connection name to search for
config_directory (str) – Directory containing config files

Returns:

Connection profile dictionary or None if not found

Return type:

Dict[str, Any] | None

Example

>>> profile = siege_utilities.find_connection_by_name("Jupyter Lab")
>>> if profile:
...     print(f"Found connection: {profile['connection_id']}")

siege_utilities.config.connections.list_connection_profiles(connection_type=None, config_directory='config')[source]

List all available connection profiles, optionally filtered by type.

Parameters:

connection_type (str) – Optional filter by connection type
config_directory (str) – Directory containing config files

Returns:

List of dictionaries with connection profile info

Return type:

List[Dict[str, Any]]

Example

>>> connections = siege_utilities.list_connection_profiles("notebook")
>>> for conn in connections:
...     print(f"{conn['name']}: {conn['status']}")

siege_utilities.config.connections.update_connection_profile(connection_id, updates, config_directory='config')[source]

Update an existing connection profile.

Parameters:

connection_id (str) – Connection ID to update
updates (Dict[str, Any]) – Dictionary of updates to apply
config_directory (str) – Directory containing config files

Raises:

FileNotFoundError – If the connection profile does not exist.
OSError – If the profile cannot be read or written.

Return type:

None

Example

>>> siege_utilities.update_connection_profile(
...     "uuid-here",
...     {"metadata": {"status": "inactive"}}
... )

siege_utilities.config.connections.verify_connection_profile(connection_id, config_directory='config')[source]

Test a connection to verify it’s working.

Parameters:

connection_id (str) – Connection ID to test
config_directory (str) – Directory containing config files

Returns:

Dictionary with test results

Return type:

Example

>>> result = siege_utilities.test_connection_profile("uuid-here")
>>> if result['success']:
...     print("Connection successful!")

siege_utilities.config.connections.get_connection_status(connection_id, config_directory='config')[source]

Get the current status and health of a connection.

Parameters:

connection_id (str) – Connection ID to check
config_directory (str) – Directory containing config files

Returns:

Dictionary with connection status information

Return type:

Example

>>> status = siege_utilities.get_connection_status("uuid-here")
>>> print(f"Status: {status['status']}, Last connected: {status['last_connected']}")

siege_utilities.config.connections.cleanup_old_connections(days_old=90, config_directory='config')[source]

Remove old, unused connection profiles.

Parameters:

days_old (int) – Remove connections older than this many days
config_directory (str) – Directory containing config files

Returns:

Number of connections removed

Return type:

int

Example

>>> removed = siege_utilities.cleanup_old_connections(30)
>>> print(f"Removed {removed} old connections")

Simple database configuration management for siege_utilities. Handles database connection settings for Spark and other uses.

siege_utilities.config.databases.create_database_config(name, connection_type, host, port, database, username, password, **kwargs)[source]

Create a database connection configuration.

Parameters:

name (str) – Friendly name for the connection
connection_type (str) – Database type (postgres, mysql, oracle, etc.)
host (str) – Database host
port (int) – Database port
database (str) – Database name
username (str) – Username
password (str) – Password
**kwargs – Additional connection parameters

Returns:

Database configuration dictionary

Return type:

Example

>>> import siege_utilities
>>> db_config = siege_utilities.create_database_config(
...     "analytics_db",
...     "postgres",
...     "localhost",
...     5432,
...     "analytics",
...     "user",
...     "password"
... )

siege_utilities.config.databases.save_database_config(config, config_directory='config')[source]

Save database configuration to JSON file.

Parameters:

config (Dict[str, Any]) – Database configuration dictionary
config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Return type:

Example

>>> db_config = create_database_config("my_db", "postgres", "localhost", 5432, "testdb", "user", "pass")
>>> file_path = siege_utilities.save_database_config(db_config)

siege_utilities.config.databases.load_database_config(db_name, config_directory='config')[source]

Load database configuration from JSON file.

Parameters:

db_name (str) – Database configuration name to load
config_directory (str) – Directory containing config files

Returns:

Database configuration dictionary.

Raises:

FileNotFoundError – If the database config file does not exist.
json.JSONDecodeError – If the file contains invalid JSON.
OSError – If the file cannot be read.

Return type:

Example

>>> db_config = siege_utilities.load_database_config("analytics_db")
>>> print(f"Database: {db_config['database']}")

siege_utilities.config.databases.get_spark_database_options(db_name, config_directory='config')[source]

Get Spark-compatible options for database connection.

Parameters:

db_name (str) – Database configuration name
config_directory (str) – Directory containing config files

Returns:

Dictionary of Spark options.

Raises:

FileNotFoundError – If the database config does not exist.
KeyError – If the config is missing required keys.

Return type:

Example

>>> spark_options = siege_utilities.get_spark_database_options("analytics_db")
>>> df = spark.read.format("jdbc").options(**spark_options).option("dbtable", "users").load()

siege_utilities.config.databases.test_database_connection(db_name, config_directory='config')[source]

Test database connection (basic connectivity check).

Parameters:

db_name (str) – Database configuration name
config_directory (str) – Directory containing config files

Returns:

True if connection successful, False otherwise

Return type:

Example

>>> if siege_utilities.test_database_connection("analytics_db"):
...     print("Database connection successful!")

siege_utilities.config.databases.list_database_configs(config_directory='config')[source]

List all available database configurations.

Parameters:: config_directory (str) – Directory containing config files
Returns:: List of dictionaries with database info
Return type:: list

Example

>>> databases = siege_utilities.list_database_configs()
>>> for db in databases:
...     print(f"{db['name']}: {db['connection_type']}")

siege_utilities.config.databases.create_spark_session_with_databases(app_name='SiegeAnalytics', database_names=None, config_directory='config')[source]

Create Spark session configured for database access.

Parameters:

app_name (str) – Spark application name
database_names (list) – List of database config names to prepare drivers for
config_directory (str) – Directory containing config files

Returns:

Configured Spark session or None if PySpark not available

Raises:

ValueError – If a database config declares a connection_type for which no JDBC driver is registered. Supported types: postgres, mysql, oracle. The error names the offending type so callers can either add a supported config or extend the dispatch.

Example

>>> spark = siege_utilities.create_spark_session_with_databases(
...     "Analytics App",
...     ["analytics_db", "staging_db"]
... )

Client configuration management for siege_utilities. Handles client profiles, contact information, and associated design artifacts.

siege_utilities.config.clients.create_client_profile(client_name, client_code, contact_info, **kwargs)[source]

Create a client profile with contact information and metadata.

Parameters:

client_name (str) – Human-readable client name
client_code (str) – Short client code (e.g., “CLIENT001”)
contact_info (Dict[str, str]) – Dictionary containing contact details
**kwargs – Additional client settings and metadata

Returns:

Dictionary with client profile configuration

Return type:

Example

>>> import siege_utilities
>>> client = siege_utilities.create_client_profile(
...     "Acme Corporation",
...     "ACME001",
...     {
...         "primary_contact": "John Doe",
...         "email": "john.doe@acme.com",
...         "phone": "+1-555-0123",
...         "address": "123 Business St, City, State"
...     },
...     industry="Technology",
...     project_count=5
... )

siege_utilities.config.clients.save_client_profile(profile, config_directory='config')[source]

Save client profile to JSON file.

Parameters:

profile (Dict[str, Any]) – Client profile dictionary
config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Return type:

Example

>>> client = create_client_profile("Test Client", "TEST001", {"email": "test@example.com"})
>>> file_path = siege_utilities.save_client_profile(client)
>>> print(f"Profile saved to: {file_path}")

siege_utilities.config.clients.load_client_profile(client_code, config_directory='config')[source]

Load client profile from JSON file.

Parameters:

client_code (str) – Client code to load
config_directory (str) – Directory containing config files

Returns:

Client profile dictionary.

Raises:

FileNotFoundError – If the profile file does not exist.
json.JSONDecodeError – If the file contains invalid JSON.
OSError – If the file cannot be read.

Return type:

Example

>>> profile = siege_utilities.load_client_profile("ACME001")
>>> print(f"Loaded: {profile['client_name']}")

siege_utilities.config.clients.update_client_profile(client_code, updates, config_directory='config')[source]

Update an existing client profile.

Parameters:

client_code (str) – Client code to update
updates (Dict[str, Any]) – Dictionary of updates to apply
config_directory (str) – Directory containing config files

Raises:

FileNotFoundError – If the client profile does not exist.
OSError – If the profile cannot be read or written.
json.JSONDecodeError – If the existing profile is corrupt.

Return type:

None

Example

>>> siege_utilities.update_client_profile(
...     "ACME001",
...     {
...         "contact_info": {"phone": "+1-555-9999"},
...         "metadata": {"project_count": 10}
...     }
... )

siege_utilities.config.clients.list_client_profiles(config_directory='config')[source]

List all available client profiles.

Parameters:: config_directory (str) – Directory containing config files
Returns:: List of dictionaries with client profile info
Return type:: List[Dict[str, Any]]

Example

>>> clients = siege_utilities.list_client_profiles()
>>> for client in clients:
...     print(f"{client['code']}: {client['name']}")

siege_utilities.config.clients.search_client_profiles(search_term, search_fields=None, config_directory='config')[source]

Search client profiles by various criteria.

Parameters:

search_term (str) – Term to search for
search_fields (List[str]) – List of fields to search in (default: all text fields)
config_directory (str) – Directory containing config files

Returns:

List of matching client profiles

Return type:

List[Dict[str, Any]]

Example

>>> results = siege_utilities.search_client_profiles(
...     "Technology",
...     ["metadata.industry", "client_name"]
... )

siege_utilities.config.clients.associate_client_with_project(client_code, project_code, config_directory='config')[source]

Associate a client with a project.

Parameters:

client_code (str) – Client code to associate
project_code (str) – Project code to associate with
config_directory (str) – Directory containing config files

Raises:

FileNotFoundError – If the client or project profile does not exist.
OSError – If profiles cannot be read or written.

Return type:

None

Example

>>> siege_utilities.associate_client_with_project("ACME001", "PROJ001")

siege_utilities.config.clients.get_client_project_associations(client_code, config_directory='config')[source]

Get all projects associated with a client.

Parameters:

client_code (str) – Client code to check
config_directory (str) – Directory containing config files

Returns:

List of associated project codes.

Raises:

FileNotFoundError – If the client profile does not exist.

Return type:

List[str]

Example

>>> projects = siege_utilities.get_client_project_associations("ACME001")
>>> print(f"Client has {len(projects)} projects")

siege_utilities.config.clients.validate_client_profile(profile)[source]

Validate a client profile and return validation results.

Parameters:: profile (Dict[str, Any]) – Client profile to validate
Returns:: Dictionary with validation results and any issues found
Return type:: Dict[str, Any]

Example

>>> profile = create_client_profile("Test", "TEST001", {"email": "test@example.com"})
>>> validation = siege_utilities.validate_client_profile(profile)
>>> if validation['is_valid']:
...     print("Profile is valid")

Simple project configuration management for siege_utilities. Handles basic project settings, directories, and metadata.

siege_utilities.config.projects.create_project_config(project_name, project_code, base_directory='projects', **kwargs)[source]

Create a simple project configuration.

SECURITY: Validates base directory path to prevent path traversal attacks.

Parameters:

project_name (str) – Human-readable project name
project_code (str) – Short project code (e.g., “PROJ001”)
base_directory (str) – Base directory for projects
**kwargs – Additional project settings

Returns:

Dictionary with project configuration

Raises:

PathSecurityError – If base_directory fails security validation

Return type:

Example

>>> import siege_utilities
>>> config = siege_utilities.create_project_config(
...     "Customer Analytics",
...     "CA001",
...     description="Analytics for customer behavior"
... )
>>> print(config['directories']['output'])
>>> create_project_config("Bad", "BAD", base_directory="../../../etc")

Security Changes:

Now validates base_directory to block path traversal
Blocks creating projects in sensitive system locations

siege_utilities.config.projects.save_project_config(config, config_directory='config')[source]

Save project configuration to JSON file.

SECURITY: Validates config directory path to prevent path traversal attacks.

Parameters:

config (Dict[str, Any]) – Project configuration dictionary
config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Raises:

PathSecurityError – If config_directory fails security validation

Return type:

Example

>>> config = create_project_config("My Project", "MP001")
>>> file_path = siege_utilities.save_project_config(config)
>>> print(f"Config saved to: {file_path}")
>>> save_project_config(config, "../../../etc")

Security Changes:

Now validates config_directory to block path traversal
Blocks saving config files to sensitive system locations

siege_utilities.config.projects.load_project_config(project_code, config_directory='config')[source]

Load project configuration from JSON file.

SECURITY: Validates config directory path to prevent path traversal attacks.

Parameters:

project_code (str) – Project code to load
config_directory (str) – Directory containing config files

Returns:

Project configuration dictionary.

Raises:

FileNotFoundError – If the project config file does not exist.
json.JSONDecodeError – If the file contains invalid JSON.
OSError – If the file cannot be read.
PathSecurityError – If config_directory fails security validation.

Return type:

Example

>>> config = siege_utilities.load_project_config("MP001")
>>> print(f"Loaded: {config['project_name']}")

siege_utilities.config.projects.setup_project_directories(config)[source]

Create the directory structure for a project.

SECURITY: Validates all directory paths in config to prevent path traversal attacks.

Parameters:

config (Dict[str, Any]) – Project configuration dictionary

Raises:

OSError – If directories cannot be created.
PathSecurityError – If any directory path fails security validation.

Return type:

None

Example

>>> config = create_project_config("My Project", "MP001")
>>> siege_utilities.setup_project_directories(config)

siege_utilities.config.projects.get_project_path(config, path_type)[source]

Get a specific path from project configuration.

Parameters:

config (Dict[str, Any]) – Project configuration dictionary
path_type (str) – Type of path (input, output, data, reports, logs, config)

Returns:

Path string or None if not found

Return type:

str | None

Example

>>> config = load_project_config("MP001")
>>> output_dir = siege_utilities.get_project_path(config, "output")
>>> print(f"Output directory: {output_dir}")

siege_utilities.config.projects.list_projects(config_directory='config')[source]

List all available project configurations.

SECURITY: Validates config directory path to prevent path traversal attacks.

Parameters:: config_directory (str) – Directory containing config files
Returns:: List of dictionaries with project info
Raises:: PathSecurityError – If config_directory fails security validation
Return type:: list

Example

>>> projects = siege_utilities.list_projects()
>>> for project in projects:
...     print(f"{project['code']}: {project['name']}")
>>> list_projects("../../../etc")

Security Changes:

Now validates config_directory to block path traversal
Blocks listing config files from sensitive system locations

siege_utilities.config.projects.update_project_config(project_code, updates, config_directory='config')[source]

Update an existing project configuration.

Parameters:

project_code (str) – Project code to update
updates (Dict[str, Any]) – Dictionary of updates to apply
config_directory (str) – Directory containing config files

Raises:

FileNotFoundError – If the project config does not exist.
OSError – If the config cannot be read or written.

Return type:

None

Example

>>> siege_utilities.update_project_config(
...     "MP001",
...     {"description": "Updated description", "settings": {"log_level": "DEBUG"}}
... )

User configuration management for siege_utilities. Handles user preferences, default settings, and personal information. Enhanced with Pydantic validation while maintaining backward compatibility.

class siege_utilities.config.user_config.UserProfile[source]

Bases: object

User profile information and preferences.

Deprecated since version Use: User from siege_utilities.config.models.actor_types instead.

username: str = ''

email: str = ''

full_name: str = ''

github_login: str = ''

organization: str = ''

preferred_download_directory: str = ''

default_output_format: str = 'pdf'

preferred_map_style: str = 'open-street-map'

default_color_scheme: str = 'YlOrRd'

default_dpi: int = 300

default_figure_size: tuple = (10, 8)

enable_logging: bool = True

log_level: str = 'INFO'

google_analytics_key: str = ''

facebook_business_key: str = ''

census_api_key: str = ''

default_database: str = 'postgresql'

postgresql_connection: str = ''

duckdb_path: str = ''

__init__(username='', email='', full_name='', github_login='', organization='', preferred_download_directory='', default_output_format='pdf', preferred_map_style='open-street-map', default_color_scheme='YlOrRd', default_dpi=300, default_figure_size=(10, 8), enable_logging=True, log_level='INFO', google_analytics_key='', facebook_business_key='', census_api_key='', default_database='postgresql', postgresql_connection='', duckdb_path='')

Parameters:

username (str)
email (str)
full_name (str)
github_login (str)
organization (str)
preferred_download_directory (str)
default_output_format (str)
preferred_map_style (str)
default_color_scheme (str)
default_dpi (int)
default_figure_size (tuple)
enable_logging (bool)
log_level (str)
google_analytics_key (str)
facebook_business_key (str)
census_api_key (str)
default_database (str)
postgresql_connection (str)
duckdb_path (str)

Return type:

None

class siege_utilities.config.user_config.UserConfigManager[source]

Bases: object

Manages user configuration and preferences. Handles user profiles, default settings, and configuration persistence.

__init__(config_dir=None)[source]

Initialize the user configuration manager.

Parameters:: config_dir (Path | None) – Directory containing user configuration files

get_user_profile()[source]

Get the current user profile.

Return type:: UserProfile

update_user_profile(**kwargs)[source]

Update user profile with new values.

Parameters:: **kwargs – Fields on UserProfile to overwrite. Accepted keys include username, email, full_name, preferred_download_directory, default_output_format, preferred_map_style, default_color_scheme, and any other UserProfile attribute. Unknown keys are logged as warnings and ignored.

get_download_directory(specific_path=None)[source]

Get the appropriate download directory.

Parameters:: specific_path (str | None) – Specific path to use instead of default
Returns:: Path object for the download directory
Return type:: Path

setup_initial_profile()[source]: Interactive setup for initial user profile.

get_api_key(service)[source]

Get API key for a specific service.

Parameters:: service (str) – Service name (e.g., ‘google_analytics’, ‘census’)
Returns:: API key string
Return type:: str

set_api_key(service, api_key)[source]

Set API key for a specific service.

Parameters:

service (str) – Service name
api_key (str) – API key value

get_database_connection(database_type=None)[source]

Get database connection string.

Parameters:: database_type (str) – Type of database (postgresql, duckdb)
Returns:: Connection string
Return type:: str

set_database_connection(database_type, connection_string)[source]

Set database connection string.

Parameters:

database_type (str) – Type of database
connection_string (str) – Connection string or path

export_config(output_path)[source]

Export user configuration to a file.

Parameters:: output_path (str) – Path to export configuration

import_config(input_path)[source]

Import user configuration from a file.

Parameters:: input_path (str) – Path to import configuration from

siege_utilities.config.user_config.get_user_config()[source]

Get or create the global user configuration manager.

Return type:: UserConfigManager

siege_utilities.config.user_config.get_download_directory(specific_path=None, client_code=None, config_dir=None)[source]

Get the appropriate download directory with hierarchical resolution.

Priority order: 1. specific_path if provided 2. Client-specific directory if client_code provided and client profile exists 3. User’s preferred download directory from profile 4. Default fallback (platform-aware: /tmp/siege_utilities/downloads

on Databricks, ~/Downloads/siege_utilities elsewhere)

On Databricks the default ~/Downloads/siege_utilities path resolves to /root/Downloads/siege_utilities which is blocked by the cluster filesystem policy. The Databricks-aware default avoids this.

Parameters:

specific_path (str | None) – Specific path override (highest priority)
client_code (str | None) – Client code for client-specific directory
config_dir (Path | None) – Configuration directory (unused, kept for compatibility)

Returns:

Path object for download directory

Raises:

OSError – If the resolved directory cannot be created or is not writable

Return type:

Path

Migration utilities for transitioning from legacy configuration system to Hydra + Pydantic.

This module provides tools to migrate existing configurations to the new system while maintaining backward compatibility.

class siege_utilities.config.enhanced_config.ConfigurationMigrator[source]

Bases: object

Handles migration from legacy configuration system to Hydra + Pydantic.

Provides utilities to migrate existing user profiles, client profiles, and other configuration data to the new system.

__init__(legacy_config_dir=None, new_config_dir=None)[source]

Initialize the migrator.

Parameters:

legacy_config_dir (Path | None) – Directory containing legacy configuration files
new_config_dir (Path | None) – Directory for new Hydra configuration files

migrate_user_profile(legacy_file=None)[source]

Migrate legacy user profile to new system.

Parameters:: legacy_file (Path | None) – Path to legacy user config file
Returns:: New UserProfile instance
Return type:: UserProfile

migrate_client_profile(legacy_file, client_code)[source]

Migrate legacy client profile to new system.

Parameters:

legacy_file (Path) – Path to legacy client config file
client_code (str) – Client code for the profile

Returns:

New ClientProfile instance

Return type:

ClientProfile

migrate_all_configurations(dry_run=False)[source]

Migrate all configurations from legacy system to new system.

Parameters:: dry_run (bool) – If True, only show what would be migrated without making changes
Returns:: Dictionary with migration results
Return type:: Dict[str, Any]

backup_legacy_configurations(backup_dir=None)[source]

Create a backup of legacy configurations before migration.

Parameters:: backup_dir (Path | None) – Directory to store backup files
Returns:: Path to backup directory
Return type:: Path

class siege_utilities.config.enhanced_config.SiegeConfig[source]

Bases: object

Legacy SiegeConfig class for backward compatibility.

__init__(config_dir=None)[source]

Parameters:: config_dir (Path | None)

get_user_profile(username)[source]

Get user profile.

Parameters:: username (str)
Return type:: UserProfile

get_client_profile(client_code)[source]

Get client profile.

Parameters:: client_code (str)
Return type:: ClientProfile

save_user_profile(profile, username)[source]

Save user profile.

Parameters:

profile (UserProfile)
username (str)

Return type:

None

save_client_profile(profile)[source]

Save client profile.

Parameters:: profile (ClientProfile)
Return type:: None

siege_utilities.config.enhanced_config.migrate_configurations(legacy_config_dir=None, dry_run=False)[source]

Convenience function to migrate all configurations.

Parameters:

legacy_config_dir (Path | None) – Directory containing legacy configuration files
dry_run (bool) – If True, only show what would be migrated without making changes

Returns:

Dictionary with migration results

Return type:

siege_utilities.config.enhanced_config.backup_and_migrate(legacy_config_dir=None, backup_dir=None)[source]

Create backup and migrate all configurations.

Parameters:

legacy_config_dir (Path | None) – Directory containing legacy configuration files
backup_dir (Path | None) – Directory to store backup files

Returns:

Dictionary with migration results including backup location

Return type:

siege_utilities.config.enhanced_config.load_user_profile(username, config_dir=None)[source]

Load user profile from YAML file (legacy compatibility).

Deprecated since version Use: HydraConfigManager.load_user() for the modern User model.

Parameters:

username (str) – Username to load
config_dir (Path | None) – Configuration directory

Returns:

UserProfile object.

Raises:

FileNotFoundError – If the profile YAML does not exist.
ValueError – If the YAML is not a mapping.
OSError – If the file cannot be read.

Return type:

UserProfile

siege_utilities.config.enhanced_config.save_user_profile(profile, username, config_dir=None)[source]

Save user profile to YAML file (legacy compatibility).

Deprecated since version Use: HydraConfigManager.save_user() for the modern User model.

Parameters:

profile (UserProfile) – UserProfile object to save
username (str) – Username to save as
config_dir (Path | None) – Configuration directory

Raises:

OSError – If the file cannot be written.
yaml.YAMLError – If serialization fails.

Return type:

None

siege_utilities.config.enhanced_config.get_download_directory(username, config_dir=None)[source]

Get download directory for user (legacy compatibility).

Parameters:

username (str) – Username
config_dir (Path | None) – Configuration directory

Returns:

Path to download directory

Raises:

FileNotFoundError – If the user profile does not exist.

Return type:

Path

siege_utilities.config.enhanced_config.load_client_profile(client_code, config_dir=None)[source]

Load client profile from YAML file (legacy compatibility).

Deprecated since version Use: HydraConfigManager.load_client() for the modern Client model.

Parameters:

client_code (str) – Client code to load
config_dir (Path | None) – Configuration directory

Returns:

ClientProfile object.

Raises:

FileNotFoundError – If the profile YAML does not exist.
ValueError – If the YAML is not a mapping.
OSError – If the file cannot be read.

Return type:

ClientProfile

siege_utilities.config.enhanced_config.save_client_profile(profile, config_dir=None)[source]

Save client profile to YAML file (legacy compatibility).

Deprecated since version Use: HydraConfigManager.save_client() for the modern Client model.

Parameters:

profile (ClientProfile) – ClientProfile object to save
config_dir (Path | None) – Configuration directory

Raises:

OSError – If the file cannot be written.
yaml.YAMLError – If serialization fails.

Return type:

None

siege_utilities.config.enhanced_config.list_client_profiles(config_dir=None)[source]

List all client profile codes (legacy compatibility).

Parameters:: config_dir (Path | None) – Configuration directory
Returns:: List of client codes
Return type:: List[str]

siege_utilities.config.enhanced_config.export_config_yaml(config_data, output_file)[source]

Export configuration data to YAML file (legacy compatibility).

Parameters:

config_data (Dict[str, Any]) – Configuration data to export
output_file (Path) – Output file path

Raises:

OSError – If the file cannot be written.
yaml.YAMLError – If serialization fails.

Return type:

None

siege_utilities.config.enhanced_config.import_config_yaml(input_file)[source]

Import configuration data from YAML file (legacy compatibility).

Parameters:

input_file (Path) – Input file path

Returns:

Configuration data.

Raises:

FileNotFoundError – If the input file does not exist.
OSError – If the file cannot be read.
yaml.YAMLError – If the YAML is malformed.

Return type:

Path configuration constants for siege_utilities. Centralized path management following the Zsh configuration pattern.

siege_utilities.config.paths.ensure_directory_exists(path)[source]

Ensure directory exists, creating it if necessary.

Parameters:: path (Path) – Path to directory
Returns:: Path object (for chaining)
Return type:: Path

siege_utilities.config.paths.get_project_path(project_name)[source]

Get path for a specific project.

Parameters:: project_name (str) – Name of the project
Returns:: Path object or None if project not found
Return type:: Path | None

siege_utilities.config.paths.get_cache_path(cache_type='default')[source]

Get cache directory path for specific cache type.

Parameters:: cache_type (str) – Type of cache (default, spark, census, nces)
Returns:: Path to cache directory
Return type:: Path

siege_utilities.config.paths.get_output_path(output_type='default')[source]

Get output directory path for specific output type.

Parameters:: output_type (str) – Type of output (default, reports, charts, maps)
Returns:: Path to output directory
Return type:: Path

siege_utilities.config.paths.get_data_path(data_type='default')[source]

Get data directory path for specific data type.

Parameters:: data_type (str) – Type of data (default, census, nces, samples)
Returns:: Path to data directory
Return type:: Path

siege_utilities.config.paths.setup_standard_directories(base_path)[source]

Set up standard directory structure in a project.

Parameters:: base_path (Path) – Base project path
Returns:: Dictionary mapping directory names to paths
Return type:: Dict[str, Path]

siege_utilities.config.paths.get_file_type(file_path)[source]

Determine file type based on extension.

Parameters:: file_path (Path) – Path to file
Returns:: File type category (data, images, documents, etc.)
Return type:: str

siege_utilities.config.paths.get_relative_to_home(path)[source]

Get path relative to user home directory.

Parameters:: path (Path) – Absolute path
Returns:: Path relative to home with ~ prefix
Return type:: str

siege_utilities.config.paths.initialize_siege_directories()[source]

Initialize all standard Siege utilities directories.

Catches PermissionError, FileNotFoundError, and OSError so that imports do not fail in containers, CI, or cross-platform environments.

Return type:: List[Path]

Simple directory management for siege_utilities. Handles directory creation, organization, and path management.

siege_utilities.config.directories.create_directory_structure(base_path, structure)[source]

Create a directory structure from a configuration dictionary.

SECURITY: Validates base path to prevent path traversal attacks.

Parameters:

base_path (str) – Base directory path
structure (Dict[str, Any]) – Dictionary defining directory structure

Returns:

Dictionary mapping directory names to created paths

Raises:

PathSecurityError – If base_path fails security validation

Return type:

Example

>>> import siege_utilities
>>> structure = {
...     "data": {"raw": {}, "processed": {}, "output": {}},
...     "reports": {},
...     "logs": {}
... }
>>> paths = siege_utilities.create_directory_structure("my_project", structure)
>>> print(paths["data/raw"])
>>> create_directory_structure("../../../etc", structure)

Security Changes:

Now validates base_path to block path traversal
Blocks creating directories in sensitive system locations

siege_utilities.config.directories.create_standard_project_structure(project_path)[source]

Create a standard project directory structure.

Parameters:: project_path (str) – Path for the project
Returns:: Dictionary mapping directory names to paths
Return type:: Dict[str, str]

Example

>>> import siege_utilities
>>> paths = siege_utilities.create_standard_project_structure("my_analytics_project")
>>> print(f"Data directory: {paths['data']}")

siege_utilities.config.directories.save_directory_config(paths, config_name, config_directory='config')[source]

Save directory configuration to JSON file.

SECURITY: Validates config_directory to prevent path traversal attacks.

Parameters:

paths (Dict[str, str]) – Dictionary of directory paths
config_name (str) – Name for the configuration
config_directory (str) – Directory to save config files

Returns:

Path to saved config file

Raises:

PathSecurityError – If config_directory fails security validation

Return type:

Example

>>> paths = create_standard_project_structure("my_project")
>>> config_file = siege_utilities.save_directory_config(paths, "my_project_dirs")
>>> save_directory_config(paths, "test", "../../etc")

Security Changes:

Now validates config_directory to block path traversal
Blocks saving configs to sensitive system locations

siege_utilities.config.directories.load_directory_config(config_name, config_directory='config')[source]

Load directory configuration from JSON file.

SECURITY: Validates config_directory to prevent path traversal attacks.

Parameters:

config_name (str) – Name of the configuration to load
config_directory (str) – Directory containing config files

Returns:

Directory configuration dictionary or None if not found

Raises:

PathSecurityError – If config_directory fails security validation

Return type:

Dict[str, Any] | None

Example

>>> dir_config = siege_utilities.load_directory_config("my_project_dirs")
>>> if dir_config:
...     print(dir_config['paths']['data'])
>>> load_directory_config("test", "../../etc")

Security Changes:

Now validates config_directory to block path traversal
Blocks loading configs from sensitive system locations

siege_utilities.config.directories.ensure_directories_exist(paths)[source]

Ensure all directories in a path configuration exist.

SECURITY: Validates each directory path to prevent path traversal attacks.

Parameters:: paths (Dict[str, str]) – Dictionary of directory paths
Returns:: True if all directories exist or were created successfully
Raises:: PathSecurityError – If any path fails security validation
Return type:: bool

Example

>>> dir_config = load_directory_config("my_project_dirs")
>>> if dir_config:
...     success = siege_utilities.ensure_directories_exist(dir_config['paths'])
>>> ensure_directories_exist({"bad": "../../etc"})

Security Changes:

Now validates each directory path to block path traversal
Blocks creating directories in sensitive system locations

siege_utilities.config.directories.get_directory_info(directory_path)[source]

Get information about a directory (size, file count, etc.).

SECURITY: Validates directory_path to prevent path traversal attacks.

Parameters:: directory_path (str) – Path to directory
Returns:: Dictionary with directory information
Raises:: PathSecurityError – If directory_path fails security validation
Return type:: Dict[str, Any]

Example

>>> info = siege_utilities.get_directory_info("my_project/data")
>>> print(f"Total files: {info['file_count']}")
>>> get_directory_info("../../etc/passwd")

Security Changes:

Now validates directory_path to block path traversal
Blocks accessing sensitive system directories

siege_utilities.config.directories.clean_empty_directories(base_path, keep_gitkeep=True)[source]

Remove empty directories (optionally keeping ones with .gitkeep).

SECURITY: Validates base_path to prevent path traversal attacks.

Parameters:

base_path (str) – Base path to start cleaning from
keep_gitkeep (bool) – If True, don’t remove directories that only contain .gitkeep

Returns:

Number of directories removed

Raises:

PathSecurityError – If base_path fails security validation

Return type:

int

Example

>>> removed = siege_utilities.clean_empty_directories("my_project/data")
>>> print(f"Removed {removed} empty directories")
>>> clean_empty_directories("../../etc")

Security Changes:

Now validates base_path to block path traversal
Blocks cleaning sensitive system directories

siege_utilities.config.directories.list_directory_configs(config_directory='config')[source]

List all available directory configurations.

SECURITY: Validates config_directory to prevent path traversal attacks.

Parameters:: config_directory (str) – Directory containing config files
Returns:: List of dictionaries with directory config info
Raises:: PathSecurityError – If config_directory fails security validation
Return type:: List[Dict[str, Any]]

Example

>>> configs = siege_utilities.list_directory_configs()
>>> for config in configs:
...     print(f"{config['name']}: {len(config['paths'])} directories")
>>> list_directory_configs("../../etc")

Security Changes:

Now validates config_directory to block path traversal
Blocks listing configs from sensitive system locations

Centralized constants for siege_utilities. Following the centralized configuration pattern from Zsh configuration.

This module provides a single source of truth for all constants used throughout the siege_utilities library, organized by domain and functionality.

siege_utilities.config.constants.get_timeout(operation_type='default')[source]

Get timeout value for specific operation types.

Different services need different timeouts based on their characteristics: - ‘default’ (30s): Standard web APIs, health checks - ‘download’ (60s): File downloads, network transfer time varies - ‘database’ (30s): Connection establishment should be fast - ‘census’ (45s): Government servers, large files, reliability critical - ‘api’ (30s): Most commercial APIs are fast - ‘extended’ (300s): Complex operations, large data processing - ‘cache’ (3600s): Data freshness vs performance balance

Parameters:: operation_type (str)
Return type:: int

siege_utilities.config.constants.get_chart_dimensions(chart_type='default')[source]

Get chart dimensions for specific chart types.

Parameters:: chart_type (str)
Return type:: tuple[float, float]

siege_utilities.config.constants.get_file_path(path_type)[source]

Get standard file paths.

Parameters:: path_type (str)
Return type:: Path

siege_utilities.config.constants.get_service_timeout(service_name)[source]

Get timeout for specific service.

Parameters:: service_name (str) – Name of the service (e.g., ‘census_api’, ‘google_analytics’)
Returns:: Timeout value in seconds, with fallback to default
Return type:: int

siege_utilities.config.constants.get_service_row_limit(service_name)[source]

Get row limit for specific service.

Parameters:: service_name (str) – Name of the service (e.g., ‘census_api’, ‘google_analytics’)
Returns:: Row limit for the service, with fallback to default
Return type:: int

Census constants — backward-compatible re-export shim.

All Census metadata now lives in census_registry.py. This module re-exports every public name so that existing from census_constants import … statements continue to work without changes.

class siege_utilities.config.census_constants.SurveyType[source]

Bases: Enum

Enumeration of Census survey types.

DECENNIAL = 'decennial'

ACS_1YR = 'acs_1yr'

ACS_3YR = 'acs_3yr'

ACS_5YR = 'acs_5yr'

CENSUS_BUSINESS = 'census_business'

POPULATION_ESTIMATES = 'population_estimates'

HOUSING_ESTIMATES = 'housing_estimates'

class siege_utilities.config.census_constants.DataReliability[source]

Bases: Enum

Enumeration of data reliability levels.

HIGH = 'high'

MEDIUM = 'medium'

LOW = 'low'

ESTIMATED = 'estimated'

class siege_utilities.config.census_constants.GeographyLevel[source]

Bases: Enum

Enumeration of Census geography levels (canonical names).

NATION = 'nation'

REGION = 'region'

DIVISION = 'division'

STATE = 'state'

COUNTY = 'county'

COUSUB = 'cousub'

PLACE = 'place'

CD = 'cd'

SLDU = 'sldu'

SLDL = 'sldl'

TRACT = 'tract'

BLOCK_GROUP = 'block_group'

BLOCK = 'block'

ZCTA = 'zcta'

CBSA = 'cbsa'

PUMA = 'puma'

VTD = 'vtd'

VTD20 = 'vtd20'

TABBLOCK20 = 'tabblock20'

TABBLOCK10 = 'tabblock10'

COUNTY_SUBDIVISION = 'cousub'

CONGRESSIONAL_DISTRICT = 'cd'

STATE_LEGISLATIVE_DISTRICT = 'sldu'

ZIP_CODE = 'zcta'

VOTING_DISTRICT = 'vtd'

siege_utilities.config.census_constants.resolve_geographic_level(level)[source]

Resolve any geographic level name variant to its canonical form.

Accepts long forms (congressional_district), short forms (cd), Census abbreviations (CD), and common aliases (zip_code → zcta).

Parameters:: level (str)
Return type:: str

siege_utilities.config.census_constants.normalize_state_identifier(state_input)[source]

Normalize a state identifier to its 2-digit zero-padded FIPS code.

Parameters:: state_input (str) – Any of: 2-digit FIPS code ("48"), 2-letter abbreviation ("TX"), or full state name ("Texas"). Case-insensitive.
Returns:: Zero-padded 2-digit FIPS code (e.g., "06" for California).
Return type:: str
Raises:: ValueError – If state_input doesn’t match any known state, territory, or DC.

Examples

>>> normalize_state_identifier("TX")
'48'
>>> normalize_state_identifier("48")
'48'
>>> normalize_state_identifier("Texas")
'48'
>>> normalize_state_identifier("DC")
'11'

siege_utilities.config.census_constants.get_tiger_url(year, state_fips, geographic_level)[source]

Generate TIGER/Line download URL.

Parameters:

year (int)
state_fips (str)
geographic_level (str)

Return type:

siege_utilities.config.census_constants.validate_geographic_level(level)[source]

Validate if geographic level is supported. Accepts any alias.

Parameters:: level (str)
Return type:: bool

siege_utilities.config.census_constants.get_fips_info(state_identifier)[source]

Return FIPS code, abbreviation, and full name for a state.

Parameters:: state_identifier (str) – Any form accepted by normalize_state_identifier().
Returns:: Keys: "fips" (zero-padded), "abbreviation", "name".
Return type:: dict

Examples

>>> get_fips_info("CA")
{'fips': '06', 'abbreviation': 'CA', 'name': 'California'}

Census Registry — single source of truth for all Census Bureau metadata.

This module consolidates Census geographic levels, variable groups, survey types, FIPS codes, TIGER URLs, and API constants that were previously scattered across census_constants.py, census_api_client.py, and census_dataset_mapper.py.

Other modules should import from here (or from the backward-compatible shim in census_constants.py).

class siege_utilities.config.census_registry.SurveyType[source]

Bases: Enum

Enumeration of Census survey types.

DECENNIAL = 'decennial'

ACS_1YR = 'acs_1yr'

ACS_3YR = 'acs_3yr'

ACS_5YR = 'acs_5yr'

CENSUS_BUSINESS = 'census_business'

POPULATION_ESTIMATES = 'population_estimates'

HOUSING_ESTIMATES = 'housing_estimates'

class siege_utilities.config.census_registry.DataReliability[source]

Bases: Enum

Enumeration of data reliability levels.

HIGH = 'high'

MEDIUM = 'medium'

LOW = 'low'

ESTIMATED = 'estimated'

class siege_utilities.config.census_registry.GeographyLevel[source]

Bases: Enum

Enumeration of Census geography levels (canonical names).

NATION = 'nation'

REGION = 'region'

DIVISION = 'division'

STATE = 'state'

COUNTY = 'county'

COUSUB = 'cousub'

PLACE = 'place'

CD = 'cd'

SLDU = 'sldu'

SLDL = 'sldl'

TRACT = 'tract'

BLOCK_GROUP = 'block_group'

BLOCK = 'block'

ZCTA = 'zcta'

CBSA = 'cbsa'

PUMA = 'puma'

VTD = 'vtd'

VTD20 = 'vtd20'

TABBLOCK20 = 'tabblock20'

TABBLOCK10 = 'tabblock10'

COUNTY_SUBDIVISION = 'cousub'

CONGRESSIONAL_DISTRICT = 'cd'

STATE_LEGISLATIVE_DISTRICT = 'sldu'

ZIP_CODE = 'zcta'

VOTING_DISTRICT = 'vtd'

siege_utilities.config.census_registry.resolve_geographic_level(level)[source]

Resolve any geographic level name variant to its canonical form.

Accepts long forms (congressional_district), short forms (cd), Census abbreviations (CD), and common aliases (zip_code → zcta).

Parameters:: level (str)
Return type:: str

siege_utilities.config.census_registry.normalize_state_identifier(state_input)[source]

Normalize a state identifier to its 2-digit zero-padded FIPS code.

Parameters:: state_input (str) – Any of: 2-digit FIPS code ("48"), 2-letter abbreviation ("TX"), or full state name ("Texas"). Case-insensitive.
Returns:: Zero-padded 2-digit FIPS code (e.g., "06" for California).
Return type:: str
Raises:: ValueError – If state_input doesn’t match any known state, territory, or DC.

Examples

>>> normalize_state_identifier("TX")
'48'
>>> normalize_state_identifier("48")
'48'
>>> normalize_state_identifier("Texas")
'48'
>>> normalize_state_identifier("DC")
'11'

siege_utilities.config.census_registry.get_tiger_url(year, state_fips, geographic_level)[source]

Generate TIGER/Line download URL.

Parameters:

year (int)
state_fips (str)
geographic_level (str)

Return type:

siege_utilities.config.census_registry.validate_geographic_level(level)[source]

Validate if geographic level is supported. Accepts any alias.

Parameters:: level (str)
Return type:: bool

siege_utilities.config.census_registry.get_fips_info(state_identifier)[source]

Return FIPS code, abbreviation, and full name for a state.

Parameters:: state_identifier (str) – Any form accepted by normalize_state_identifier().
Returns:: Keys: "fips" (zero-padded), "abbreviation", "name".
Return type:: dict

Examples

>>> get_fips_info("CA")
{'fips': '06', 'abbreviation': 'CA', 'name': 'California'}

NCES (National Center for Education Statistics) constants for siege_utilities. Centralized configuration for NCES data sources and urbanicity calculations.

This module supports the planned NCES urbanicity integration described in IMPROVEMENTS.md.

siege_utilities.config.nces_constants.get_locale_category(locale_code)[source]

Get the 4-category locale type from numeric code.

Parameters:: locale_code (int) – NCES numeric locale code (11-43)
Returns:: Locale category (city, suburban, town, rural)
Raises:: ValueError – If locale code is not recognized
Return type:: str

siege_utilities.config.nces_constants.get_locale_subcategory(locale_code)[source]

Get the 12-category locale subcategory from numeric code.

Parameters:: locale_code (int) – NCES numeric locale code (11-43)
Returns:: Locale subcategory (e.g., ‘city_large’, ‘rural_remote’)
Return type:: str

siege_utilities.config.nces_constants.classify_urbanicity(population, distance_to_urban)[source]

Classify urbanicity based on population and distance to urban areas.

Parameters:

population (int) – Population of the area
distance_to_urban (float) – Distance to nearest urbanized area (miles)

Returns:

Urbanicity classification string

Return type:

siege_utilities.config.nces_constants.get_nces_download_url(data_type, year)[source]

Generate NCES download URL.

Parameters:

data_type (str) – Type of NCES data (locale_boundaries, school_locations, etc.)
year (int) – Data year

Returns:

Complete URL for NCES data download

Raises:

ValueError – If parameters are invalid

Return type:

siege_utilities.config.nces_constants.validate_locale_code(code)[source]

Validate if NCES locale code is recognized.

Parameters:: code (int)
Return type:: bool

siege_utilities.config.nces_constants.get_urbanicity_info(locale_code)[source]

Get comprehensive urbanicity information for a locale code.

Parameters:: locale_code (int) – NCES numeric locale code
Returns:: Dictionary with category, subcategory, and classification info
Return type:: Dict[str, Any]

DataSource registry for managing external data provider configurations.

Provides a YAML-backed registry of data sources with built-in entries for FEC, Census, NLRB, and NCES. Supports filtering by type and jurisdiction.

class siege_utilities.config.data_source_registry.DataSourceRegistry[source]

Bases: object

Registry of external data sources.

Loads built-in sources on init and optionally overlays a YAML file for user-defined sources and overrides.

__init__(config_path=None)[source]

Parameters:: config_path (Path | None)

register_source(source)[source]

Register or update a data source.

Parameters:: source (DataSource)
Return type:: None

get_source(source_id)[source]

Get a data source by ID.

Parameters:: source_id (str)
Return type:: DataSource | None

list_sources(include_planned=True)[source]

List all registered sources, optionally filtering out planned ones.

Parameters:: include_planned (bool)
Return type:: List[DataSource]

list_by_type(source_type)[source]

List sources filtered by type.

Parameters:: source_type (DataSourceType)
Return type:: List[DataSource]

list_by_jurisdiction(level=None, state_abbreviation=None)[source]

List sources filtered by jurisdiction level and/or state.

Parameters:

level (JurisdictionLevel | None)
state_abbreviation (str | None)

Return type:

List[DataSource]

list_active()[source]

List only active sources.

Return type:: List[DataSource]

register_credential(credential)[source]

Register a credential for a source.

Parameters:: credential (SourceCredential)
Return type:: None

get_credential(source_id, environment='production')[source]

Get credential for a source in a given environment.

Parameters:

source_id (str)
environment (str)

Return type:

SourceCredential | None

load_from_yaml(path)[source]

Load sources and credentials from a YAML file.

Parameters:: path (Path)
Return type:: None

save_to_yaml(path)[source]

Save all sources and credentials to a YAML file.

Parameters:: path (Path)
Return type:: None

Google Account registry for managing multiple Google accounts.

Provides a JSON-backed registry with default selection, filtering, and persistence. Follows the DataSourceRegistry pattern.

class siege_utilities.config.google_account_registry.GoogleAccountRegistry[source]

Bases: object

Manages multiple GoogleAccounts with default selection and persistence.

__init__(config_path=None)[source]

Parameters:: config_path (Path | None)

register(account)[source]

Register or update a Google account.

Enforces at most one default: if account.is_default is True, all other accounts are demoted.

Parameters:: account (GoogleAccount)
Return type:: None

remove(google_account_id)[source]

Remove a Google account by ID. Returns True if found.

Parameters:: google_account_id (str)
Return type:: bool

get(google_account_id)[source]

Get a Google account by ID.

Parameters:: google_account_id (str)
Return type:: GoogleAccount | None

get_by_email(email)[source]

Get the first Google account matching email.

Parameters:: email (str)
Return type:: GoogleAccount | None

get_default()[source]

Get the default account, or None.

Return type:: GoogleAccount | None

set_default(google_account_id)[source]

Set an account as the default. Raises ValueError if not found.

Parameters:: google_account_id (str)
Return type:: None

list_accounts(status=None)[source]

List all accounts, optionally filtered by status.

Parameters:: status (GoogleAccountStatus | None)
Return type:: List[GoogleAccount]

list_active()[source]

List only active accounts.

Return type:: List[GoogleAccount]

list_by_type(account_type)[source]

List accounts filtered by type.

Parameters:: account_type (GoogleAccountType)
Return type:: List[GoogleAccount]

load(path)[source]

Load accounts from a JSON file.

Parameters:: path (Path)
Return type:: None

save(path)[source]

Save all accounts to a JSON file.

Parameters:: path (Path)
Return type:: None

siege_utilities.config.google_account_registry.list_google_accounts_for_owner(owner_id, config_directory='config')[source]

Discover Google account profiles for an owner via file glob.

Parameters:

owner_id (str)
config_directory (str)

Return type:

List[GoogleAccount]

siege_utilities.config.google_account_registry.save_google_account_profile(account, owner_id, config_directory='config')[source]

Save a single GoogleAccount to a per-owner JSON file. Returns the path.

Parameters:

account (GoogleAccount)
owner_id (str)
config_directory (str)

Return type:

siege_utilities.config.google_account_registry.load_google_account_profile(owner_id, google_account_id, config_directory='config')[source]

Load a single GoogleAccount from a per-owner JSON file.

Parameters:

owner_id (str)
google_account_id (str)
config_directory (str)

Return type:

GoogleAccount | None

siege_utilities.config.google_account_registry.migrate_single_account(oauth_integration, google_account_id, email, display_name)[source]

Convert an existing OAuthIntegration into a GoogleAccount.

Copies scopes and the integration name as a reference.

Parameters:

google_account_id (str)
email (str)
display_name (str)

Return type:

GoogleAccount

Hydra Configuration Manager for siege_utilities.

This module provides a unified interface for loading and managing configurations using Hydra with Pydantic validation.

class siege_utilities.config.hydra_manager.HydraConfigManager[source]

Bases: object

Hydra Configuration Manager with Pydantic validation.

Provides a unified interface for loading configurations using Hydra and validating them with Pydantic models.

__init__(config_dir=None)[source]

Initialize the Hydra configuration manager.

Parameters:: config_dir (Path | None) – Directory containing Hydra configuration files

load_config(config_name, overrides=None)[source]

Load configuration with Hydra.

Parameters:

config_name (str) – Name of the configuration to load
overrides (List[str] | None) – List of override strings (e.g., [“+client=client_a”])

Returns:

Dictionary containing the loaded configuration

Return type:

load_user_profile(client_code=None)[source]

Load user profile with optional client overrides.

Parameters:: client_code (str | None) – Optional client code for client-specific overrides
Returns:: Validated UserProfile instance
Return type:: UserProfile

load_client_profile(client_code)[source]

Load client profile with client-specific overrides.

Parameters:: client_code (str) – Client code for the profile to load
Returns:: Validated ClientProfile instance
Return type:: ClientProfile

load_database_connections(client_code=None)[source]

Load database connections with optional client overrides.

Parameters:: client_code (str | None) – Optional client code for client-specific connections
Returns:: List of validated DatabaseConnection instances
Return type:: List[DatabaseConnection]

load_social_media_accounts(client_code=None)[source]

Load social media accounts with optional client overrides.

Parameters:: client_code (str | None) – Optional client code for client-specific accounts
Returns:: List of validated SocialMediaAccount instances
Return type:: List[SocialMediaAccount]

load_branding_config(client_code=None)[source]

Load branding configuration with optional client overrides.

Parameters:: client_code (str | None) – Optional client code for client-specific branding
Returns:: Validated BrandingConfig instance
Return type:: BrandingConfig

load_user(person_id, profiles_dir=None)[source]

Load a modern User from YAML file.

Parameters:

person_id (str) – The person_id of the user to load.
profiles_dir (Path | None) – Directory containing user YAML files. Defaults to config_dir/profiles/users.

Returns:

User instance.

Raises:

FileNotFoundError – If the user profile YAML file does not exist.
OSError – If the file cannot be read.
yaml.YAMLError – If the YAML is malformed.
ValueError – If the data cannot be parsed into a User.

Return type:

User

load_client(client_code, profiles_dir=None)[source]

Load a modern Client from YAML file.

Parameters:

client_code (str) – The client_code to load.
profiles_dir (Path | None) – Directory containing client YAML files. Defaults to config_dir/profiles/clients.

Returns:

Client instance.

Raises:

FileNotFoundError – If the client profile YAML file does not exist.
OSError – If the file cannot be read.
yaml.YAMLError – If the YAML is malformed.
ValueError – If the data cannot be parsed into a Client.

Return type:

Client

save_user(user, profiles_dir=None)[source]

Save a modern User to YAML file.

Parameters:

user (User) – User instance to save.
profiles_dir (Path | None) – Directory to save the YAML file. Defaults to config_dir/profiles/users.

Raises:

OSError – If the file cannot be written.
yaml.YAMLError – If serialization fails.

Return type:

None

save_client(client, profiles_dir=None)[source]

Save a modern Client to YAML file.

Parameters:

client (Client) – Client instance to save.
profiles_dir (Path | None) – Directory to save the YAML file. Defaults to config_dir/profiles/clients.

Raises:

OSError – If the file cannot be written.
yaml.YAMLError – If serialization fails.

Return type:

None

cleanup()[source]: Clean up Hydra instance.

Migration utilities for transitioning from legacy configuration system to Hydra + Pydantic.

This module provides tools to migrate existing configurations to the new system while maintaining backward compatibility.

class siege_utilities.config.migration.ConfigurationMigrator[source]

Bases: object

Handles migration from legacy configuration system to Hydra + Pydantic.

Provides utilities to migrate existing user profiles, client profiles, and other configuration data to the new system.

__init__(legacy_config_dir=None, new_config_dir=None)[source]

Initialize the migrator.

Parameters:

legacy_config_dir (Path | None) – Directory containing legacy configuration files
new_config_dir (Path | None) – Directory for new Hydra configuration files

migrate_user_profile(legacy_file=None)[source]

Migrate legacy user profile to new system.

Parameters:: legacy_file (Path | None) – Path to legacy user config file
Returns:: New UserProfile instance
Return type:: UserProfile

migrate_client_profile(legacy_file, client_code)[source]

Migrate legacy client profile to new system.

Parameters:

legacy_file (Path) – Path to legacy client config file
client_code (str) – Client code for the profile

Returns:

New ClientProfile instance

Return type:

ClientProfile

migrate_all_configurations(dry_run=False)[source]

Migrate all configurations from legacy system to new system.

Parameters:: dry_run (bool) – If True, only show what would be migrated without making changes
Returns:: Dictionary with migration results
Return type:: Dict[str, Any]

backup_legacy_configurations(backup_dir=None)[source]

Create a backup of legacy configurations before migration.

Parameters:: backup_dir (Path | None) – Directory to store backup files
Returns:: Path to backup directory
Return type:: Path

siege_utilities.config.migration.migrate_configurations(legacy_config_dir=None, dry_run=False)[source]

Convenience function to migrate all configurations.

Parameters:

legacy_config_dir (Path | None) – Directory containing legacy configuration files
dry_run (bool) – If True, only show what would be migrated without making changes

Returns:

Dictionary with migration results

Return type:

siege_utilities.config.migration.backup_and_migrate(legacy_config_dir=None, backup_dir=None)[source]

Create backup and migrate all configurations.

Parameters:

legacy_config_dir (Path | None) – Directory containing legacy configuration files
backup_dir (Path | None) – Directory to store backup files

Returns:

Dictionary with migration results including backup location

Return type:

Models

Enhanced Pydantic models for siege_utilities configuration system.

This module provides comprehensive data validation models for all configurable entities in the siege_utilities system.

class siege_utilities.config.models.UserProfile[source]

Bases: BaseModel

Enhanced user profile with comprehensive validation.

Provides type-safe, validated user configuration with detailed field constraints and business logic validation.

validate_figure_size(): Validate figure size constraints.

validate_username(): Validate username format if provided.

validate_email(): Validate email format if provided.

validate_github_login(): Validate GitHub login format if provided.

validate_api_key(): Validate API key format if provided.

validate_download_directory(): Ensure download directory is a Path object.

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.ClientProfile[source]

Bases: BaseModel

Enhanced client profile with comprehensive validation.

Provides type-safe, validated client configuration with detailed field constraints and business logic validation.

contact_info: ContactInfo

branding_config: BrandingConfig

report_preferences: ReportPreferences

notes: str | None = None

validate_client_code(): Validate client code format and uniqueness.

validate_database_connections(): Validate database connections for uniqueness.

validate_social_media_accounts(): Validate social media accounts for uniqueness.

get_active_database_connections()[source]

Get all active database connections.

Return type:: List[DatabaseConnection]

get_active_social_media_accounts()[source]

Get all active social media accounts.

Return type:: List[SocialMediaAccount]

get_database_connection_by_name(name)[source]

Get database connection by name.

Parameters:: name (str)
Return type:: DatabaseConnection | None

get_social_media_account_by_platform(platform)[source]

Get social media account by platform.

Parameters:: platform (str)
Return type:: SocialMediaAccount | None

add_database_connection(connection)[source]

Add a new database connection.

Parameters:: connection (DatabaseConnection)
Return type:: None

add_social_media_account(account)[source]

Add a new social media account.

Parameters:: account (SocialMediaAccount)
Return type:: None

update_branding_config(branding)[source]

Update branding configuration.

Parameters:: branding (BrandingConfig)
Return type:: None

update_report_preferences(preferences)[source]

Update report preferences.

Parameters:: preferences (ReportPreferences)
Return type:: None

get_summary()[source]

Get client profile summary.

Return type:: dict

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.ContactInfo[source]

Bases: BaseModel

Contact information with validation.

phone: str | None = None

address: str | None = None

website: str | None = None

linkedin: str | None = None

validate_email(): Validate email format if provided.

validate_phone(): Validate phone number format.

validate_website(): Validate website URL format.

validate_linkedin(): Validate LinkedIn URL format.

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.DatabaseConnection[source]

Bases: BaseModel

Database connection configuration with comprehensive validation.

Provides type-safe, validated database connection settings with detailed field constraints and security validation.

validate_host(): Validate host format.

validate_password(): Validate password strength.

validate_port(): Validate port number.

validate_database_name(): Validate database name.

get_connection_string()[source]

Generate connection string for the database.

Return type:: str

class siege_utilities.config.models.SocialMediaAccount[source]

Bases: BaseModel

Social media account configuration with comprehensive validation.

Provides type-safe, validated social media account settings with platform-specific validation and security constraints.

refresh_token: str | None = None

api_version: str | None = None

rate_limit_remaining: int | None = None

last_updated: datetime | None = None

validate_access_token(): Validate access token format.

validate_account_id(): Validate account ID based on platform.

validate_api_version(): Validate API version based on platform.

get_api_base_url()[source]

Get the API base URL for the platform.

Return type:: str

get_auth_headers()[source]

Get authentication headers for API requests.

Return type:: dict

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.BrandingConfig[source]

Bases: BaseModel

Branding configuration with comprehensive validation.

Provides type-safe, validated branding settings with color format validation and design constraints.

logo_path: Path | None = None

logo_width: int | None = None

logo_height: int | None = None

validate_color_contrast(): Validate color format and provide warnings for potential contrast issues.

validate_font_name(): Validate font name format.

validate_logo_path(): Validate logo path format.

validate_logo_dimensions(): Validate logo dimensions.

get_color_scheme()[source]

Get the complete color scheme as a dictionary.

Return type:: dict

get_typography_scheme()[source]

Get the complete typography scheme as a dictionary.

Return type:: dict

get_layout_scheme()[source]

Get the complete layout scheme as a dictionary.

Return type:: dict

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.ReportPreferences[source]

Bases: BaseModel

Report preferences with comprehensive validation.

Provides type-safe, validated report generation settings with format constraints and layout validation.

watermark_text: str | None = None

validate_sections(): Validate report sections.

validate_watermark(): Validate watermark settings.

validate_chart_dpi(): Validate chart DPI based on quality setting.

get_export_settings()[source]

Get export settings as a dictionary.

Return type:: dict

get_chart_settings()[source]

Get chart settings as a dictionary.

Return type:: dict

get_layout_settings()[source]

Get layout settings as a dictionary.

Return type:: dict

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.JurisdictionLevel[source]

Bases: str, Enum

Level of government a data source covers.

FEDERAL = 'federal'

STATE = 'state'

COUNTY = 'county'

MUNICIPAL = 'municipal'

TRIBAL = 'tribal'

__new__(value)

class siege_utilities.config.models.Jurisdiction[source]

Bases: BaseModel

Geographic/governmental jurisdiction for a data source.

level: JurisdictionLevel

validate_state_fips_format()

validate_county_fips_format()

validate_jurisdiction_consistency()

class siege_utilities.config.models.DataSourceType[source]

Bases: str, Enum

Category of data a source provides.

CAMPAIGN_FINANCE = 'campaign_finance'

VOTER_FILE = 'voter_file'

BOUNDARY = 'boundary'

DEMOGRAPHIC = 'demographic'

EDUCATION = 'education'

LABOR = 'labor'

ELECTION_ADMIN = 'election_admin'

__new__(value)

class siege_utilities.config.models.DataSourceStatus[source]

Bases: str, Enum

Lifecycle status of a data source integration.

ACTIVE = 'active'

PLANNED = 'planned'

DEPRECATED = 'deprecated'

UNAVAILABLE = 'unavailable'

__new__(value)

class siege_utilities.config.models.DataSource[source]

Bases: BaseModel

Registry entry for an external data provider.

source_type: DataSourceType

jurisdiction: Jurisdiction

status: DataSourceStatus = 'planned'

auth_required: bool = False

notes: str | None = None

validate_auth_type()

class siege_utilities.config.models.SourceCredential[source]

Bases: BaseModel

Per-source authentication credential reference (not the secret itself).

last_verified: datetime | None = None

class siege_utilities.config.models.GoogleAccount[source]

Bases: BaseModel

A Google account with credential references and scope tracking.

For OAuth accounts, oauth_integration_name references an OAuthIntegration on the owning Person. For service accounts, service_account_ref points to a 1Password item or file path.

validate_email(): Validate email format.

validate_google_account_id(): Validate account ID is non-empty after stripping.

validate_credential_refs()

Ensure the credential reference matches account_type.

Return type:: GoogleAccount

is_active()[source]

Check if account is active.

Return type:: bool

update_last_used()[source]

Update the last-used timestamp.

Return type:: None

get_info()[source]

Get account info (excluding sensitive credential data).

Return type:: Dict[str, Any]

class siege_utilities.config.models.GoogleAccountType[source]

Bases: str, Enum

Type of Google account authentication.

OAUTH = 'oauth'

SERVICE_ACCOUNT = 'service_account'

__new__(value)

class siege_utilities.config.models.GoogleAccountStatus[source]

Bases: str, Enum

Status of a Google account.

ACTIVE = 'active'

REVOKED = 'revoked'

EXPIRED = 'expired'

PENDING = 'pending'

__new__(value)

Specific actor types that extend the base Person model.

class siege_utilities.config.models.actor_types.User[source]

Bases: Person

User actor - Siege/Masai team members and internal users.

Extends Person with user-specific fields and capabilities.

validate_username(): Validate username format.

validate_github_login(): Validate GitHub login format if provided.

validate_role(): Validate user role.

validate_permissions(): Validate permissions list.

has_permission(permission)[source]

Check if user has a specific permission.

Parameters:: permission (str)
Return type:: bool

add_permission(permission)[source]

Add a permission to this user.

Parameters:: permission (str)
Return type:: None

remove_permission(permission)[source]

Remove a permission from this user.

Parameters:: permission (str)
Return type:: None

get_source_credential(source_id)[source]

Get the credential reference for a data source.

Parameters:: source_id (str)
Return type:: str | None

set_source_credential(source_id, credential_ref)[source]

Set a credential reference for a data source.

Parameters:

source_id (str)
credential_ref (str)

Return type:

None

has_jurisdiction_access(jurisdiction_name)[source]

Check if user is authorized for a jurisdiction (empty list = all).

Parameters:: jurisdiction_name (str)
Return type:: bool

assign_client(client_code)[source]

Assign a client to this user.

Parameters:: client_code (str)
Return type:: None

unassign_client(client_code)[source]

Unassign a client from this user. Returns True if removed.

Parameters:: client_code (str)
Return type:: bool

set_primary_client(client_code)[source]

Set the primary client for this user. Must be in assigned_clients.

Parameters:: client_code (str)
Return type:: None

get_assigned_clients()[source]

Get list of assigned client codes.

Return type:: List[str]

has_client(client_code)[source]

Check if user is assigned to a client.

Parameters:: client_code (str)
Return type:: bool

validate_assigned_clients(): Validate assigned clients list.

validate_primary_client(): Validate primary client is in assigned list.

class siege_utilities.config.models.actor_types.Client[source]

Bases: Person

Client actor - External client organizations.

Extends Person with client-specific fields and capabilities.

coerce_branding_config(): Accept Dict input and coerce to BrandingConfig for backward compatibility.

coerce_report_preferences(): Accept Dict input and coerce to ReportPreferences for backward compatibility.

validate_client_code(): Validate client code format and uniqueness.

increment_project_count()[source]

Increment the project count.

Return type:: None

decrement_project_count()[source]

Decrement the project count.

Return type:: None

is_active_client()[source]

Check if client is active.

Return type:: bool

operates_in_jurisdiction(jurisdiction_name)[source]

Check if client operates in a jurisdiction (empty list = all).

Parameters:: jurisdiction_name (str)
Return type:: bool

has_source_enabled(source_id)[source]

Check if a data source is enabled for this client.

Parameters:: source_id (str)
Return type:: bool

assign_user(person_id)[source]

Assign a user to this client.

Parameters:: person_id (str)
Return type:: None

unassign_user(person_id)[source]

Unassign a user from this client. Returns True if removed.

Parameters:: person_id (str)
Return type:: bool

set_primary_user(person_id)[source]

Set the primary user for this client. Must be in assigned_users.

Parameters:: person_id (str)
Return type:: None

get_assigned_users()[source]

Get list of assigned user person_ids.

Return type:: List[str]

has_user(person_id)[source]

Check if a user is assigned to this client.

Parameters:: person_id (str)
Return type:: bool

validate_assigned_users(): Validate assigned users list.

validate_primary_user(): Validate primary user is in assigned list.

class siege_utilities.config.models.actor_types.Collaborator[source]

Bases: Person

Collaborator actor - External collaborators (e.g., Tony from Masai).

Extends Person with collaboration-specific fields and capabilities.

validate_external_organization(): Validate external organization name.

validate_access_expires(): Validate access expiration date.

validate_allowed_services(): Validate allowed services list.

validate_restricted_credentials(): Validate restricted credentials list.

is_access_expired()[source]

Check if collaboration access has expired.

Return type:: bool

is_collaboration_active()[source]

Check if collaboration is active (not expired and person is active).

Return type:: bool

can_access_service(service)[source]

Check if collaborator can access a specific service.

Parameters:: service (str)
Return type:: bool

can_access_credential(credential_name)[source]

Check if collaborator can access a specific credential.

Parameters:: credential_name (str)
Return type:: bool

accept_invitation()[source]

Accept collaboration invitation.

Return type:: None

extend_access(days)[source]

Extend collaboration access by specified days.

Parameters:: days (int)
Return type:: None

class siege_utilities.config.models.actor_types.Organization[source]

Bases: BaseModel

Organization model for companies (Siege, Masai, Hillcrest).

validate_primary_email(): Validate primary email format.

validate_members(): Validate members list.

add_member(person_id)[source]

Add a member to this organization.

Parameters:: person_id (str)
Return type:: None

remove_member(person_id)[source]

Remove a member from this organization.

Parameters:: person_id (str)
Return type:: None

set_primary_contact(person_id)[source]

Set the primary contact for this organization.

Parameters:: person_id (str)
Return type:: None

get_summary()[source]

Get organization summary information.

Return type:: Dict[str, Any]

to_dict(exclude_sensitive=False)[source]

Convert to dictionary, optionally stripping sensitive fields.

Parameters:: exclude_sensitive (bool)
Return type:: dict

to_yaml(path=None, exclude_sensitive=False)[source]

Serialize to YAML string, optionally writing to a file.

Parameters:

path (Path | None)
exclude_sensitive (bool)

Return type:

classmethod from_yaml(yaml_str_or_path)[source]

Deserialize from a YAML string or file path.

Parameters:: yaml_str_or_path (str | Path)
Return type:: Organization

class siege_utilities.config.models.actor_types.Collaboration[source]

Bases: BaseModel

Collaboration model for joint projects between organizations.

validate_participants(): Validate participant lists.

validate_end_date(): Validate end date.

add_organization(org_id)[source]

Add an organization to this collaboration.

Parameters:: org_id (str)
Return type:: None

add_client(client_id)[source]

Add a client to this collaboration.

Parameters:: client_id (str)
Return type:: None

add_participant(person_id)[source]

Add a participant to this collaboration.

Parameters:: person_id (str)
Return type:: None

is_active()[source]

Check if collaboration is active.

Return type:: bool

get_summary()[source]

Get collaboration summary information.

Return type:: Dict[str, Any]

to_dict(exclude_sensitive=False)[source]

Convert to dictionary, optionally stripping sensitive fields.

Parameters:: exclude_sensitive (bool)
Return type:: dict

to_yaml(path=None, exclude_sensitive=False)[source]

Serialize to YAML string, optionally writing to a file.

Parameters:

path (Path | None)
exclude_sensitive (bool)

Return type:

classmethod from_yaml(yaml_str_or_path)[source]

Deserialize from a YAML string or file path.

Parameters:: yaml_str_or_path (str | Path)
Return type:: Collaboration

Branding configuration model with comprehensive validation.

class siege_utilities.config.models.branding_config.BrandingConfig[source]

Bases: BaseModel

Branding configuration with comprehensive validation.

Provides type-safe, validated branding settings with color format validation and design constraints.

logo_path: Path | None = None

logo_width: int | None = None

logo_height: int | None = None

validate_color_contrast(): Validate color format and provide warnings for potential contrast issues.

validate_font_name(): Validate font name format.

validate_logo_path(): Validate logo path format.

validate_logo_dimensions(): Validate logo dimensions.

get_color_scheme()[source]

Get the complete color scheme as a dictionary.

Return type:: dict

get_typography_scheme()[source]

Get the complete typography scheme as a dictionary.

Return type:: dict

get_layout_scheme()[source]

Get the complete layout scheme as a dictionary.

Return type:: dict

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

Enhanced client profile model with comprehensive validation.

class siege_utilities.config.models.client_profile.ContactInfo[source]

Bases: BaseModel

Contact information with validation.

phone: str | None = None

address: str | None = None

website: str | None = None

linkedin: str | None = None

validate_email(): Validate email format if provided.

validate_phone(): Validate phone number format.

validate_website(): Validate website URL format.

validate_linkedin(): Validate LinkedIn URL format.

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.client_profile.ClientProfile[source]

Bases: BaseModel

Enhanced client profile with comprehensive validation.

Provides type-safe, validated client configuration with detailed field constraints and business logic validation.

contact_info: ContactInfo

branding_config: BrandingConfig

report_preferences: ReportPreferences

notes: str | None = None

validate_client_code(): Validate client code format and uniqueness.

validate_database_connections(): Validate database connections for uniqueness.

validate_social_media_accounts(): Validate social media accounts for uniqueness.

get_active_database_connections()[source]

Get all active database connections.

Return type:: List[DatabaseConnection]

get_active_social_media_accounts()[source]

Get all active social media accounts.

Return type:: List[SocialMediaAccount]

get_database_connection_by_name(name)[source]

Get database connection by name.

Parameters:: name (str)
Return type:: DatabaseConnection | None

get_social_media_account_by_platform(platform)[source]

Get social media account by platform.

Parameters:: platform (str)
Return type:: SocialMediaAccount | None

add_database_connection(connection)[source]

Add a new database connection.

Parameters:: connection (DatabaseConnection)
Return type:: None

add_social_media_account(account)[source]

Add a new social media account.

Parameters:: account (SocialMediaAccount)
Return type:: None

update_branding_config(branding)[source]

Update branding configuration.

Parameters:: branding (BrandingConfig)
Return type:: None

update_report_preferences(preferences)[source]

Update report preferences.

Parameters:: preferences (ReportPreferences)
Return type:: None

get_summary()[source]

Get client profile summary.

Return type:: dict

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

Credential management model with comprehensive validation.

class siege_utilities.config.models.credential.CredentialType[source]

Bases: str, Enum

Types of credentials supported.

API_KEY = 'api_key'

OAUTH_TOKEN = 'oauth_token'

USERNAME_PASSWORD = 'username_password'

SSH_KEY = 'ssh_key'

CERTIFICATE = 'certificate'

SECRET = 'secret'

__new__(value)

class siege_utilities.config.models.credential.CredentialStatus[source]

Bases: str, Enum

Status of credentials.

ACTIVE = 'active'

EXPIRED = 'expired'

REVOKED = 'revoked'

PENDING = 'pending'

__new__(value)

class siege_utilities.config.models.credential.Credential[source]

Bases: BaseModel

Credential management with comprehensive validation.

Provides type-safe, validated credential storage with detailed field constraints and security validation.

validate_name(): Validate credential name.

validate_service(): Validate service name.

validate_password(): Validate password strength if provided.

validate_api_key(): Validate API key format if provided.

validate_expires_at(): Validate expiration date.

is_expired()[source]

Check if credential is expired.

Return type:: bool

is_valid()[source]

Check if credential is valid (active and not expired).

Return type:: bool

update_last_used()[source]

Update the last used timestamp.

Return type:: None

get_credential_data()[source]

Get credential data as dictionary (excluding sensitive fields for logging).

Return type:: Dict[str, Any]

get_connection_info()[source]

Get connection information for this credential.

Return type:: Dict[str, Any]

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

class siege_utilities.config.models.credential.OnePasswordCredential[source]

Bases: BaseModel

Specialized credential model for 1Password integration.

validate_1password_id(): Validate 1Password ID format.

get_1password_reference()[source]

Get 1Password reference string.

Return type:: str

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

Data source models for multi-source data ingestion.

Provides Jurisdiction, DataSource, and SourceCredential models for tracking external data providers and their authentication.

class siege_utilities.config.models.data_sources.JurisdictionLevel[source]

Bases: str, Enum

Level of government a data source covers.

FEDERAL = 'federal'

STATE = 'state'

COUNTY = 'county'

MUNICIPAL = 'municipal'

TRIBAL = 'tribal'

__new__(value)

class siege_utilities.config.models.data_sources.DataSourceType[source]

Bases: str, Enum

Category of data a source provides.

CAMPAIGN_FINANCE = 'campaign_finance'

VOTER_FILE = 'voter_file'

BOUNDARY = 'boundary'

DEMOGRAPHIC = 'demographic'

EDUCATION = 'education'

LABOR = 'labor'

ELECTION_ADMIN = 'election_admin'

__new__(value)

class siege_utilities.config.models.data_sources.DataSourceStatus[source]

Bases: str, Enum

Lifecycle status of a data source integration.

ACTIVE = 'active'

PLANNED = 'planned'

DEPRECATED = 'deprecated'

UNAVAILABLE = 'unavailable'

__new__(value)

class siege_utilities.config.models.data_sources.Jurisdiction[source]

Bases: BaseModel

Geographic/governmental jurisdiction for a data source.

level: JurisdictionLevel

validate_state_fips_format()

validate_county_fips_format()

validate_jurisdiction_consistency()

class siege_utilities.config.models.data_sources.DataSource[source]

Bases: BaseModel

Registry entry for an external data provider.

source_type: DataSourceType

jurisdiction: Jurisdiction

status: DataSourceStatus = 'planned'

auth_required: bool = False

notes: str | None = None

validate_auth_type()

class siege_utilities.config.models.data_sources.SourceCredential[source]

Bases: BaseModel

Per-source authentication credential reference (not the secret itself).

last_verified: datetime | None = None

Database connection model with comprehensive validation.

class siege_utilities.config.models.database_connection.DatabaseConnection[source]

Bases: BaseModel

Database connection configuration with comprehensive validation.

Provides type-safe, validated database connection settings with detailed field constraints and security validation.

validate_host(): Validate host format.

validate_password(): Validate password strength.

validate_port(): Validate port number.

validate_database_name(): Validate database name.

get_connection_string()[source]

Generate connection string for the database.

Return type:: str

Bulk import/export for Person/Actor entity collections.

Provides functions to serialize all entity types (users, clients, collaborators, organizations, collaborations) to a single YAML document and deserialize back.

siege_utilities.config.models.export.export_entities(users=None, clients=None, collaborators=None, organizations=None, collaborations=None, path=None, exclude_sensitive=False)[source]

Export all entity types to a single YAML document.

Parameters:

users (List[User] | None) – List of User instances to export.
clients (List[Client] | None) – List of Client instances to export.
collaborators (List[Collaborator] | None) – List of Collaborator instances to export.
organizations (List[Organization] | None) – List of Organization instances to export.
collaborations (List[Collaboration] | None) – List of Collaboration instances to export.
path (Path | None) – Optional file path to write the YAML to.
exclude_sensitive (bool) – If True, redact sensitive credential data.

Returns:

YAML string of all exported entities.

Return type:

siege_utilities.config.models.export.import_entities(yaml_str_or_path)[source]

Import entities from a YAML document.

Parameters:: yaml_str_or_path (str | Path) – YAML string or path to a YAML file.
Returns:: Dictionary with keys ‘users’, ‘clients’, ‘collaborators’, ‘organizations’, ‘collaborations’, each containing a list of the corresponding model instances. Also includes ‘version’ and ‘exported_at’ metadata.
Return type:: Dict[str, Any]

Google Account model for multi-account management.

Represents a Google account (OAuth or Service Account) with metadata for credential resolution, scope tracking, and default selection.

class siege_utilities.config.models.google_account.GoogleAccountType[source]

Bases: str, Enum

Type of Google account authentication.

OAUTH = 'oauth'

SERVICE_ACCOUNT = 'service_account'

__new__(value)

class siege_utilities.config.models.google_account.GoogleAccountStatus[source]

Bases: str, Enum

Status of a Google account.

ACTIVE = 'active'

REVOKED = 'revoked'

EXPIRED = 'expired'

PENDING = 'pending'

__new__(value)

class siege_utilities.config.models.google_account.GoogleAccount[source]

Bases: BaseModel

A Google account with credential references and scope tracking.

For OAuth accounts, oauth_integration_name references an OAuthIntegration on the owning Person. For service accounts, service_account_ref points to a 1Password item or file path.

validate_email(): Validate email format.

validate_google_account_id(): Validate account ID is non-empty after stripping.

validate_credential_refs()

Ensure the credential reference matches account_type.

Return type:: GoogleAccount

is_active()[source]

Check if account is active.

Return type:: bool

update_last_used()[source]

Update the last-used timestamp.

Return type:: None

get_info()[source]

Get account info (excluding sensitive credential data).

Return type:: Dict[str, Any]

OAuth integration model with comprehensive validation.

class siege_utilities.config.models.oauth_integration.OAuthProvider[source]

Bases: str, Enum

Supported OAuth providers.

GOOGLE = 'google'

MICROSOFT = 'microsoft'

GITHUB = 'github'

LINKEDIN = 'linkedin'

FACEBOOK = 'facebook'

TWITTER = 'twitter'

SALESFORCE = 'salesforce'

SLACK = 'slack'

ZOOM = 'zoom'

HUBSPOT = 'hubspot'

ZOHO = 'zoho'

DYNAMICS = 'dynamics'

CUSTOM = 'custom'

__new__(value)

class siege_utilities.config.models.oauth_integration.OAuthScope[source]

Bases: str, Enum

Common OAuth scopes.

READ = 'read'

WRITE = 'write'

ADMIN = 'admin'

PROFILE = 'profile'

EMAIL = 'email'

ANALYTICS = 'analytics'

FILES = 'files'

CALENDAR = 'calendar'

CONTACTS = 'contacts'

__new__(value)

class siege_utilities.config.models.oauth_integration.OAuthIntegration[source]

Bases: BaseModel

OAuth integration configuration with comprehensive validation.

Provides type-safe, validated OAuth settings with detailed field constraints and security validation.

validate_name(): Validate integration name.

validate_service(): Validate service name.

validate_redirect_uri(): Validate redirect URI format.

validate_oauth_credentials(): Validate OAuth credentials.

validate_expires_at(): Validate token expiration date.

is_token_expired()[source]

Check if access token is expired.

Return type:: bool

needs_refresh()[source]

Check if token needs refresh based on threshold.

Return type:: bool

is_valid()[source]

Check if integration is valid (active and has valid token).

Return type:: bool

update_last_used()[source]

Update the last used timestamp.

Return type:: None

update_tokens(access_token, refresh_token=None, expires_in=None)[source]

Update OAuth tokens.

Parameters:

access_token (str)
refresh_token (str | None)
expires_in (int | None)

Return type:

None

get_authorization_url(state=None)[source]

Generate OAuth authorization URL.

Parameters:: state (str | None)
Return type:: str

get_integration_info()[source]

Get integration information (excluding sensitive data).

Return type:: Dict[str, Any]

model_config = {'extra': 'forbid', 'json_schema_serialization_mode': 'json', 'validate_assignment': True}

Base Person model with comprehensive validation and credential management.

class siege_utilities.config.models.person.Person[source]

Bases: BaseModel

Base person/actor model with common capabilities.

Provides the foundation for all person types (User, Client, Collaborator) with shared credential management, relationships, and common fields.

validate_email(): Validate email format.

validate_phone(): Validate phone number format.

validate_website(): Validate website URL format.

validate_linkedin(): Validate LinkedIn URL format.

validate_organizations(): Validate organizations list.

validate_collaborations(): Validate collaborations list.

add_credential(credential)[source]

Add a credential to this person.

Parameters:: credential (Credential)
Return type:: None

remove_credential(credential_name)[source]

Remove a credential by name.

Parameters:: credential_name (str)
Return type:: bool

get_credential(credential_name)[source]

Get a credential by name.

Parameters:: credential_name (str)
Return type:: Credential | None

get_credentials_for_service(service)[source]

Get all credentials for a specific service.

Parameters:: service (str)
Return type:: List[Credential]

get_valid_credentials()[source]

Get all valid (active and not expired) credentials.

Return type:: List[Credential]

add_oauth_integration(integration)[source]

Add an OAuth integration to this person.

Parameters:: integration (OAuthIntegration)
Return type:: None

remove_oauth_integration(integration_name)[source]

Remove an OAuth integration by name.

Parameters:: integration_name (str)
Return type:: bool

get_oauth_integration(integration_name)[source]

Get an OAuth integration by name.

Parameters:: integration_name (str)
Return type:: OAuthIntegration | None

get_valid_oauth_integrations()[source]

Get all valid OAuth integrations.

Return type:: List[OAuthIntegration]

add_database_connection(connection)[source]

Add a database connection to this person.

Parameters:: connection (DatabaseConnection)
Return type:: None

remove_database_connection(connection_name)[source]

Remove a database connection by name.

Parameters:: connection_name (str)
Return type:: bool

get_database_connection(connection_name)[source]

Get a database connection by name.

Parameters:: connection_name (str)
Return type:: DatabaseConnection | None

add_onepassword_credential(credential)[source]

Add a 1Password credential reference to this person.

Parameters:: credential (OnePasswordCredential)
Return type:: None

remove_onepassword_credential(credential_name, vault_id)[source]

Remove a 1Password credential reference.

Parameters:

credential_name (str)
vault_id (str)

Return type:

add_google_account(account)[source]

Add a Google account to this person.

Parameters:: account (GoogleAccount)
Return type:: None

remove_google_account(google_account_id)[source]

Remove a Google account by ID.

Parameters:: google_account_id (str)
Return type:: bool

get_google_account(google_account_id)[source]

Get a Google account by ID.

Parameters:: google_account_id (str)
Return type:: GoogleAccount | None

get_google_account_by_email(email)[source]

Get a Google account by email address.

Parameters:: email (str)
Return type:: GoogleAccount | None

get_default_google_account()[source]

Get the default Google account.

Return type:: GoogleAccount | None

set_default_google_account(google_account_id)[source]

Set a Google account as the default.

Parameters:: google_account_id (str)
Return type:: None

list_google_accounts(active_only=False)[source]

List Google accounts, optionally filtering to active only.

Parameters:: active_only (bool)
Return type:: List[GoogleAccount]

add_organization(org_id)[source]

Add an organization to this person.

Parameters:: org_id (str)
Return type:: None

remove_organization(org_id)[source]

Remove an organization from this person.

Parameters:: org_id (str)
Return type:: None

set_primary_organization(org_id)[source]

Set the primary organization for this person.

Parameters:: org_id (str)
Return type:: None

add_collaboration(collab_id)[source]

Add a collaboration to this person.

Parameters:: collab_id (str)
Return type:: None

remove_collaboration(collab_id)[source]

Remove a collaboration from this person.

Parameters:: collab_id (str)
Return type:: None

get_summary()[source]

Get person summary information.

Return type:: Dict[str, Any]

is_active()[source]

Check if person is active.

Return type:: bool

has_organization(org_id)[source]

Check if person belongs to an organization.

Parameters:: org_id (str)
Return type:: bool

has_collaboration(collab_id)[source]

Check if person participates in a collaboration.

Parameters:: collab_id (str)
Return type:: bool

get_onepassword_credential(credential_name)[source]

Get a 1Password credential by name.

Parameters:: credential_name (str)
Return type:: OnePasswordCredential | None

get_onepassword_credentials_for_service(service)[source]

Get all 1Password credentials for a specific service.

Parameters:: service (str)
Return type:: List[OnePasswordCredential]

get_onepassword_credentials_for_vault(vault_id)[source]

Get all 1Password credentials for a specific vault.

Parameters:: vault_id (str)
Return type:: List[OnePasswordCredential]

has_onepassword_credential(credential_name)[source]

Check if person has a specific 1Password credential.

Parameters:: credential_name (str)
Return type:: bool

has_onepassword_credentials_for_service(service)[source]

Check if person has 1Password credentials for a specific service.

Parameters:: service (str)
Return type:: bool

get_onepassword_services()[source]

Get list of services that have 1Password credentials.

Return type:: List[str]

get_onepassword_vaults()[source]

Get list of 1Password vaults used by this person.

Return type:: List[str]

get_credential_coverage()[source]

Get comprehensive credential coverage analysis.

Return type:: Dict[str, Dict[str, Any]]

get_security_recommendations()[source]

Get security recommendations based on credential analysis.

Return type:: List[str]

to_dict(exclude_sensitive=False)[source]

Convert to dictionary, optionally stripping sensitive credential data.

Parameters:: exclude_sensitive (bool)
Return type:: dict

to_yaml(path=None, exclude_sensitive=False)[source]

Serialize to YAML string, optionally writing to a file.

Parameters:

path (Path | None)
exclude_sensitive (bool)

Return type: