Analytics
Analytics connectors: Google Analytics, Snowflake, data.world, Facebook Business.
Analytics Module — lazy-loaded.
Provides analytics integration: Google Analytics, Facebook Business, Snowflake, Data.world connectors, and Google Workspace write APIs (Sheets, Docs, Slides).
- class siege_utilities.analytics.GoogleAnalyticsConnector[source]
Bases:
objectGoogle Analytics connection and data retrieval manager.
- __init__(client_id=None, client_secret=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', auth_method='oauth2', service_account_data=None)[source]
Initialize Google Analytics connector.
- Parameters:
client_id (str) – OAuth2 client ID from Google Cloud Console (for OAuth2)
client_secret (str) – OAuth2 client secret from Google Cloud Console (for OAuth2)
redirect_uri (str) – OAuth2 redirect URI (for OAuth2)
auth_method (str) – Authentication method (“oauth2” or “service_account”)
service_account_data (Dict[str, Any]) – Service account credentials dict (for service account auth)
- authenticate_service_account()[source]
Authenticate with Google Analytics using service account. Based on working implementation from GA project.
- Raises:
ValueError – If no service account data is available.
google.auth.exceptions.* – On authentication failure.
OSError – If temp file operations fail.
- Return type:
None
- get_ga4_data(property_id, start_date, end_date, metrics, dimensions=None, row_limit=100000)[source]
Retrieve data from Google Analytics 4.
- Parameters:
property_id (str) – GA4 property ID
start_date (str) – Start date as
YYYY-MM-DD(also accepts the GA4 relative keywordstoday,yesterday, orNdaysAgo). Interpreted in the property’s configured timezone, not the caller’s local time — set this in the GA admin UI.end_date (str) – End date, same format / TZ rules as
start_date.row_limit (int) – Maximum rows to retrieve
- Returns:
Pandas DataFrame with GA4 data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate or recognized GA4 relative keyword. Previously, malformed dates were forwarded raw to the GA API and returned an opaque 400 hours later in a batch run.- Return type:
- get_ua_data(view_id, start_date, end_date, metrics, dimensions=None, max_results=100000)[source]
Retrieve data from Universal Analytics.
- Parameters:
- Returns:
Pandas DataFrame with UA data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate or recognized GA relative keyword.- Return type:
- save_as_pandas(df, output_path, format='parquet')[source]
Save DataFrame as Pandas format.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to save
output_path (str) – Output file path
format (str) – Output format (parquet, csv, excel, etc.)
- Raises:
ValueError – If the format is unsupported.
OSError – If the file cannot be written.
- Return type:
None
- save_as_spark(df, output_path, spark_session=None)[source]
Save DataFrame as Spark DataFrame and optionally to storage.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to convert
output_path (str) – Output path for Spark DataFrame
spark_session (pyspark.sql.SparkSession | None) – Optional SparkSession (will create if not provided)
- Raises:
ImportError – If PySpark is not available.
OSError – If the output path cannot be written.
- Return type:
None
- siege_utilities.analytics.create_ga_account_profile(client_id, ga_property_id, account_type='ga4', credentials_file=None)[source]
Create a Google Analytics account profile linked to a client.
- Parameters:
- Returns:
GA account profile dictionary
- Return type:
- siege_utilities.analytics.save_ga_account_profile(profile, config_directory='config')[source]
Save GA account profile to JSON file.
- siege_utilities.analytics.load_ga_account_profile(account_id, config_directory='config')[source]
Load GA account profile from JSON file.
- siege_utilities.analytics.list_ga_accounts_for_client(client_id, config_directory='config')[source]
List all GA accounts associated with a client.
- siege_utilities.analytics.batch_retrieve_ga_data(client_id, start_date, end_date, metrics, dimensions=None, output_format='pandas', output_directory='output/ga_data')[source]
Batch retrieve GA data for all accounts associated with a client.
- Parameters:
- Returns:
Dictionary with retrieval results
- Return type:
- siege_utilities.analytics.create_ga_connector_with_service_account(service_account_data=None)[source]
Create a GoogleAnalyticsConnector using service account authentication.
- class siege_utilities.analytics.FacebookBusinessConnector[source]
Bases:
objectFacebook Business connection and data retrieval manager.
- __init__(access_token, app_id=None, app_secret=None)[source]
Initialize Facebook Business connector.
- get_ad_insights(ad_account_id, start_date, end_date, fields=None, breakdowns=None)[source]
Retrieve ad insights from Facebook Ads.
- Parameters:
- Returns:
Pandas DataFrame with ad insights
- Return type:
- get_page_insights(page_id, start_date, end_date, metrics=None)[source]
Retrieve page insights from Facebook Pages.
- get_business_insights(business_id, start_date, end_date, metrics=None)[source]
Retrieve business insights from Facebook Business Manager.
- save_as_pandas(df, output_path, format='parquet')[source]
Save DataFrame as Pandas format.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to save
output_path (str) – Output file path
format (str) – Output format (parquet, csv, excel, etc.)
- Raises:
ValueError – If the format is unsupported.
OSError – If the file cannot be written.
- Return type:
None
- save_as_spark(df, output_path, spark_session=None)[source]
Save DataFrame as Spark DataFrame and optionally to storage.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to convert
output_path (str) – Output path for Spark DataFrame
spark_session (pyspark.sql.SparkSession | None) – Optional SparkSession (will create if not provided)
- Raises:
ImportError – If PySpark is not available.
OSError – If the output path cannot be written.
- Return type:
None
- siege_utilities.analytics.create_facebook_account_profile(client_id, fb_account_id, account_type='ad_account', access_token=None)[source]
Create a Facebook account profile linked to a client.
- siege_utilities.analytics.save_facebook_account_profile(profile, config_directory='config')[source]
Save Facebook account profile to JSON file.
- siege_utilities.analytics.load_facebook_account_profile(account_id, config_directory='config')[source]
Load Facebook account profile from JSON file.
- siege_utilities.analytics.list_facebook_accounts_for_client(client_id, config_directory='config')[source]
List all Facebook accounts associated with a client.
- siege_utilities.analytics.batch_retrieve_facebook_data(client_id, start_date, end_date, data_types=None, output_format='pandas', output_directory='output/facebook_data')[source]
Batch retrieve Facebook data for all accounts associated with a client.
- Parameters:
client_id (str) – Client ID to retrieve data for
start_date (str) – Start date for data retrieval
end_date (str) – End date for data retrieval
data_types (List[str]) – Types of data to retrieve (ads, pages, business)
output_format (str) – Output format (pandas, spark, or both)
output_directory (str) – Directory to save output files
- Returns:
Dictionary with retrieval results
- Return type:
- class siege_utilities.analytics.SnowflakeConnector[source]
Bases:
objectSnowflake data warehouse connector with advanced features.
- __init__(account, user, password=None, warehouse=None, database=None, schema=None, role=None, config_file=None)[source]
Initialize Snowflake connector.
- Parameters:
account (str) – Snowflake account identifier
user (str) – Snowflake username
password (str | None) – Snowflake password (optional if using config file)
warehouse (str | None) – Default warehouse to use
database (str | None) – Default database to use
schema (str | None) – Default schema to use
role (str | None) – Default role to use
config_file (str | Path | None) – Path to configuration file
- connect()[source]
Establish connection to Snowflake.
- Raises:
ConnectionError – If the connection attempt fails.
- Return type:
None
- execute_query(query, params=None)[source]
Execute a SQL query.
- Parameters:
- Returns:
Query results as list of tuples
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If query execution fails.
- Return type:
- execute_ddl(ddl_statement)[source]
Execute DDL statements (CREATE, ALTER, DROP, etc.).
- Parameters:
ddl_statement (str) – DDL SQL statement
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If DDL execution fails.
- Return type:
None
- upload_dataframe(df, table_name, database=None, schema=None, auto_create_table=True, overwrite=False)[source]
Upload pandas DataFrame to Snowflake table.
- Parameters:
df (pd.DataFrame) – Pandas DataFrame to upload
table_name (str) – Target table name
database (str | None) – Target database (uses default if not specified)
schema (str | None) – Target schema (uses default if not specified)
auto_create_table (bool) – Whether to automatically create table if it doesn’t exist
overwrite (bool) – Whether to overwrite existing table
- Raises:
ImportError – If pandas is not available.
ConnectionError – If not connected and connection fails.
RuntimeError – If the upload fails.
- Return type:
None
- download_dataframe(query, params=None)[source]
Download data from Snowflake as pandas DataFrame.
- Parameters:
- Returns:
Pandas DataFrame with query results
- Raises:
ImportError – If pandas is not available.
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
pd.DataFrame
- get_table_info(table_name, database=None, schema=None)[source]
Get information about a Snowflake table.
- Parameters:
- Returns:
Dictionary with table information
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- list_tables(database=None, schema=None)[source]
List all tables in a database/schema.
- Parameters:
- Returns:
List of table names
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- siege_utilities.analytics.get_snowflake_connector(config_file=None)[source]
Get Snowflake connector instance from configuration.
- Parameters:
- Return type:
- siege_utilities.analytics.upload_to_snowflake(df, table_name, config_file=None, **kwargs)[source]
Convenience function to upload DataFrame to Snowflake.
- siege_utilities.analytics.download_from_snowflake(query, config_file=None, **kwargs)[source]
Convenience function to download DataFrame from Snowflake.
- siege_utilities.analytics.execute_snowflake_query(query, config_file=None, **kwargs)[source]
Convenience function to execute Snowflake query.
- class siege_utilities.analytics.DataDotWorldConnector[source]
Bases:
objectData.world connector with advanced data discovery and access features.
- search_datasets(query, limit=50, owner=None, tags=None)[source]
Search for datasets on data.world.
- Parameters:
- Returns:
List of dataset information dictionaries
- Raises:
ConnectionError – If the API request fails.
- Return type:
- get_dataset_info(dataset_id)[source]
Get detailed information about a specific dataset.
- Parameters:
dataset_id (str) – Dataset identifier (format: owner/dataset)
- Returns:
Dictionary with detailed dataset information
- Raises:
ConnectionError – If the API request fails.
- Return type:
- download_file(dataset_id, file_name, output_path=None)[source]
Download a specific file from a dataset.
- Parameters:
- Returns:
Path to the downloaded file
- Raises:
ConnectionError – If the download fails.
OSError – If writing to disk fails.
- Return type:
- load_dataset_as_dataframe(dataset_id, file_name=None)[source]
Load a dataset file directly as a pandas DataFrame.
- Parameters:
- Returns:
Pandas DataFrame with the data
- Raises:
ImportError – If pandas is not available.
ConnectionError – If the API request fails.
- Return type:
pd.DataFrame
- query_dataset(dataset_id, query, query_type='sql')[source]
Execute a query against a dataset.
- Parameters:
- Returns:
Pandas DataFrame with query results
- Raises:
ImportError – If pandas is not available.
ValueError – If query_type is invalid.
ConnectionError – If the query fails.
- Return type:
pd.DataFrame
- get_dataset_schema(dataset_id)[source]
Get the schema information for a dataset.
- Parameters:
dataset_id (str) – Dataset identifier (format: owner/dataset)
- Returns:
Dictionary with schema information
- Raises:
ConnectionError – If the API request fails.
- Return type:
- create_dataset(owner, dataset_id, title, description, tags=None, license='Public Domain', visibility='OPEN')[source]
Create a new dataset on data.world.
- Parameters:
- Returns:
owner/dataset_id)
- Return type:
Created dataset ID (format
- Raises:
ValueError – If API token is not configured.
ConnectionError – If the API request fails.
- upload_file(dataset_id, file_path, file_name=None)[source]
Upload a file to an existing dataset.
- Parameters:
- Raises:
ValueError – If API token is not configured.
FileNotFoundError – If the local file does not exist.
ConnectionError – If the upload fails.
- Return type:
None
- update_dataset(dataset_id, title=None, description=None, tags=None, license=None)[source]
Update an existing dataset.
- Parameters:
- Raises:
ValueError – If API token is not configured or no updates provided.
ConnectionError – If the API request fails.
- Return type:
None
- delete_dataset(dataset_id)[source]
Delete a dataset.
- Parameters:
dataset_id (str) – Dataset identifier (format: owner/dataset)
- Raises:
ValueError – If API token is not configured.
ConnectionError – If the API request fails.
- Return type:
None
- siege_utilities.analytics.get_datadotworld_connector(config_file=None)[source]
Get Data.world connector instance from configuration.
- Parameters:
- Return type:
- siege_utilities.analytics.search_datasets(query, config_file=None, **kwargs)[source]
Standalone function to search for datasets.
- siege_utilities.analytics.list_datasets(limit=50, config_file=None, **kwargs)[source]
Standalone function to list available datasets.
- siege_utilities.analytics.search_datadotworld_datasets(query, config_file=None, **kwargs)[source]
Convenience function to search for datasets.
- siege_utilities.analytics.load_datadotworld_dataset(dataset_id, config_file=None, **kwargs)[source]
Convenience function to load a dataset as DataFrame.
- siege_utilities.analytics.query_datadotworld_dataset(dataset_id, query, config_file=None, **kwargs)[source]
Convenience function to query a dataset.
- Parameters:
- Return type:
pd.DataFrame
- class siege_utilities.analytics.GoogleWorkspaceClient[source]
Bases:
objectAuthenticated client that builds Google API service objects.
Do not instantiate directly — use the
from_oauth()orfrom_service_account()class methods.- classmethod from_oauth(client_id, client_secret, token_file=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', scopes=None)[source]
Authenticate via OAuth2 interactive flow.
If token_file exists and contains a valid/refreshable token, no browser interaction is needed.
- classmethod from_1password(item_title='Google OAuth Client - siege_utilities', vault=None, account='TLTQ3ANAABGCNEK7KIAOTDNK2Q', token_file=None, scopes=None)[source]
Authenticate using credentials stored in a 1Password Document item.
Auto-detects whether the JSON document is an OAuth client secret (has
"installed"or"web"key) or a service account key (has"type": "service_account"), and routes to the appropriate auth flow.For OAuth: runs the installed-app flow (browser required on first use; cached token reused afterward).
For service account: authenticates server-to-server with no browser interaction.
- Parameters:
item_title (str) – Title of the 1Password Document item.
vault (str | None) – 1Password vault name.
account (str | None) – 1Password account shorthand or UUID.
token_file (str | Path | None) – Path to cache the OAuth token (ignored for service accounts). Default:
~/.siege/tokens/workspace_token.json.scopes (List[str] | None) – OAuth scopes (defaults to
WORKSPACE_SCOPES).
- Returns:
Authenticated GoogleWorkspaceClient.
- Return type:
- classmethod from_service_account(service_account_data=None, service_account_file=None, scopes=None)[source]
Authenticate via service account credentials.
Provide service_account_data (dict) or service_account_file (path). If neither is given, attempts to fetch from 1Password via
CredentialManager.get_google_service_account_from_1password().
- classmethod from_credentials(credentials)[source]
Wrap an already-authenticated
google.auth.credentials.Credentials.- Return type:
- classmethod from_account(account, person=None, scopes=None)[source]
Build a client from a
GoogleAccount.For OAuth accounts the method looks for a cached token_file first, then resolves oauth_integration_name from person to get client_id/client_secret for the OAuth flow.
For service accounts the method resolves service_account_ref via
CredentialManager(1Password) or treats it as a file path.- Parameters:
account (GoogleAccount)
person (Optional[Person])
scopes (Optional[List[str]])
- Return type:
- classmethod from_registry(registry, google_account_id=None, person=None, scopes=None)[source]
Build a client from a registry’s default or specified account.
If google_account_id is given, looks it up directly; otherwise uses the registry’s default account.
- Parameters:
registry (GoogleAccountRegistry)
google_account_id (Optional[str])
person (Optional[Person])
scopes (Optional[List[str]])
- Return type:
- property credentials
- batch_update_presentation(presentation_id, requests)[source]
Execute a batch of Slides API requests.
- static spreadsheet_url(spreadsheet_id)[source]
Return the live Google Sheets URL for a spreadsheet ID.
- static presentation_url(presentation_id)[source]
Return the live Google Slides URL for a presentation ID.
- static file_url(file_id, mime_type=None)[source]
Return the live Google Drive URL for a file ID.
If mime_type is provided, returns the appropriate editor URL. Otherwise returns the generic Drive file URL.
- copy_file(file_id, title=None)[source]
Copy a Drive file (spreadsheet, doc, presentation) and return the new ID.
Share a Drive file with a user.
- siege_utilities.analytics.create_spreadsheet(client, title, sheet_names=None, folder_id=None)[source]
Create a new Google Spreadsheet and return its ID.
- Parameters:
- Returns:
The spreadsheet ID string.
- Return type:
- siege_utilities.analytics.write_values(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Write a 2-D list of values to a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.append_rows(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Append rows after existing data in a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.read_values(client, spreadsheet_id, range_)[source]
Read values from a range.
- siege_utilities.analytics.write_dataframe(client, spreadsheet_id, df, sheet_name='Sheet1', include_header=True, start_cell='A1')[source]
Write a pandas DataFrame to a sheet.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.read_dataframe(client, spreadsheet_id, range_='Sheet1', has_header=True)[source]
Read a range from a spreadsheet into a pandas DataFrame.
- siege_utilities.analytics.add_sheet(client, spreadsheet_id, title)[source]
Add a new tab/sheet to an existing spreadsheet.
- siege_utilities.analytics.get_spreadsheet_metadata(client, spreadsheet_id)[source]
Fetch spreadsheet metadata (title, sheets, locale, etc.).
- siege_utilities.analytics.copy_spreadsheet(client, spreadsheet_id, title=None)[source]
Copy an entire spreadsheet via the Drive API.
- siege_utilities.analytics.create_presentation(client, title, folder_id=None)[source]
Create a new Google Slides presentation and return its ID.
- siege_utilities.analytics.get_presentation(client, presentation_id)[source]
Fetch full presentation metadata.
- siege_utilities.analytics.copy_presentation(client, presentation_id, title=None)[source]
Copy an entire presentation via the Drive API.
- siege_utilities.analytics.add_blank_slide(client, presentation_id, layout='BLANK', insertion_index=None)[source]
Add a blank slide to a presentation.
- Parameters:
- Returns:
The new slide’s object ID.
- Return type:
- siege_utilities.analytics.create_textbox(client, presentation_id, slide_id, text, left=100, top=100, width=400, height=50)[source]
Create a text box on a slide and populate it with text.
Dimensions are in EMU (English Metric Units). 1 inch = 914400 EMU. For convenience, this function accepts points (1 pt = 12700 EMU) but the values are treated as points internally.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
presentation_id (str) – Target presentation ID.
slide_id (str) – Slide object ID.
text (str) – Text to put in the box.
left (float) – Position in points from top-left.
top (float) – Position in points from top-left.
width (float) – Dimensions in points.
height (float) – Dimensions in points.
- Returns:
The textbox object ID.
- Return type:
- siege_utilities.analytics.create_document(client, title, folder_id=None)[source]
Create a new Google Doc and return its document ID.
- siege_utilities.analytics.get_document(client, document_id)[source]
Fetch full document metadata and content structure.
- siege_utilities.analytics.copy_document(client, document_id, title=None)[source]
Copy a document via the Drive API.
- siege_utilities.analytics.read_document_text(client, document_id)[source]
Read the full plain-text content of a document.
Extracts text from all structural elements (paragraphs, tables, etc.).
- siege_utilities.analytics.insert_paragraph(client, document_id, text, index=1, heading=None)[source]
Insert a paragraph with optional heading style.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
document_id (str) – Target document ID.
text (str) – Paragraph text (newline is appended automatically).
index (int) – Character index (1-based).
heading (str | None) – Named style (
"HEADING_1"through"HEADING_6","TITLE","SUBTITLE", orNonefor normal text).
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.insert_table(client, document_id, rows, cols, index=1)[source]
Insert an empty table at the given index.
- siege_utilities.analytics.replace_text(client, document_id, find, replace_with, match_case=True)[source]
Find and replace text throughout a document.
- class siege_utilities.analytics.VistaSocialConnector[source]
Bases:
objectBearer-token authenticated client for the Vista Social REST API.
Usage:
from siege_utilities.analytics.vista_social import VistaSocialConnector client = VistaSocialConnector(api_token=os.environ["VISTA_SOCIAL_TOKEN"]) accounts = client.list_accounts() analytics = client.get_account_analytics( account_id=accounts[0]["id"], start_date="2026-01-01", end_date="2026-01-31", )
All methods raise:
VistaSocialAuthErroron 401/403,VistaSocialRateLimitErroron 429,VistaSocialErroron other non-2xx responses or network failures.
The connector follows the project’s failure-mode discipline (see
docs/FAILURE_MODES.md): no method silently returns an empty dict on failure. Every failure mode is either a typed exception or a None return with a documented meaning.- __init__(api_token, *, base_url='https://api.vistasocial.com', timeout=30, retry_attempts=3, retry_backoff=1.5)[source]
- list_accounts()[source]
Return the list of accounts this token can access.
Wraps
GET /v1/accounts. Returns the decodeddatalist from the response, or an empty list if the API returned nothing. Raises on auth / rate-limit / network failures (no silent empty-on-error).
- list_profiles(account_id)[source]
Return social profiles connected to account_id.
Wraps
GET /v1/accounts/{account_id}/profiles.
- get_account_analytics(account_id, *, start_date, end_date, metrics=None)[source]
Fetch aggregate analytics for account_id over a date range.
- Parameters:
account_id (str) – Vista Social account id (string, opaque).
start_date (str) – ISO-8601 date (
YYYY-MM-DD), inclusive.end_date (str) – ISO-8601 date, inclusive.
metrics (List[str] | None) – Optional list of metric names (e.g.
["impressions", "engagements", "reach", "follower_count"]). If omitted, Vista Social returns its default metric set.
- Return type:
Wraps
GET /v1/accounts/{account_id}/analytics. Returns the decoded JSON body verbatim – Vista Social returns a structured document with metric breakdowns; consumers can extract what they need.
- list_posts(account_id, *, start_date=None, end_date=None, limit=100)[source]
Return posts for account_id with optional date filtering.
Wraps
GET /v1/accounts/{account_id}/posts. Pagination is intentionally simple here – consumers that need full enumeration should call repeatedly with adjusted date ranges or extend this connector with cursor handling once the real API spec is known.
- class siege_utilities.analytics.VistaSocialResponse[source]
Bases:
objectStructured response from a Vista Social API call.
The connector returns this rather than the raw
requests.Responseso consumers don’t need to know about the underlying HTTP library.- data
Parsed JSON body. Empty dict if the response had no body (e.g. 204 No Content) or wasn’t JSON-decodable.
- Type:
Mapping[str, Any]
- headers
Response headers – useful for inspecting rate-limit counters (
X-RateLimit-Remaining,Retry-After).
- exception siege_utilities.analytics.VistaSocialError[source]
Bases:
RuntimeErrorBase class for all Vista Social connector failures.
- exception siege_utilities.analytics.VistaSocialAuthError[source]
Bases:
VistaSocialErrorAuthentication failed (401/403). Token is missing, wrong, or expired.
- exception siege_utilities.analytics.VistaSocialRateLimitError[source]
Bases:
VistaSocialErrorAPI rate limit hit (429). Caller can back off and retry later.
- class siege_utilities.analytics.Dimension[source]
Bases:
objectA scoring dimension with a name and weight (0-100).
- class siege_utilities.analytics.ThresholdConfig[source]
Bases:
objectTier thresholds for score classification.
Scores >= green_min are Green, >= yellow_min are Yellow, else Red.
- class siege_utilities.analytics.HealthScorer[source]
Bases:
objectWeighted multi-dimensional entity scorer.
- Parameters:
dimensions (list[Dimension]) – Scoring dimensions. Weights must sum to 100.
default_thresholds (ThresholdConfig, optional) – Default tier thresholds. Can be overridden per-segment.
segment_thresholds (dict[str, ThresholdConfig], optional) – Per-segment threshold overrides.
- Raises:
ValueError – If dimensions is empty or weights don’t sum to 100.
- __init__(dimensions, default_thresholds=None, segment_thresholds=None)[source]
- Parameters:
default_thresholds (ThresholdConfig | None)
segment_thresholds (dict[str, ThresholdConfig] | None)
- Return type:
None
- score(entity_id, scores, *, segment=None, previous_score=None)[source]
Score an entity across all dimensions.
- Parameters:
- Return type:
- Raises:
ValueError – If scores dict is missing dimensions or values are out of range.
- score_batch(entities)[source]
Score multiple entities.
Each dict must have ‘entity_id’ and ‘scores’ keys. Optional: ‘segment’, ‘previous_score’.
- Parameters:
entities (list[dict]) – List of entity dicts with keys: entity_id, scores, and optionally segment, previous_score.
- Return type:
- Raises:
ValueError – If any entity dict is missing required keys.
- class siege_utilities.analytics.SignalType[source]
Bases:
EnumTypes of risk signals.
- DECLINE = 'decline'
- THRESHOLD = 'threshold'
- ABSENCE = 'absence'
- ANOMALY = 'anomaly'
- class siege_utilities.analytics.RiskTier[source]
Bases:
EnumRisk classification tiers, ordered by severity.
- CRITICAL = 'Critical'
- HIGH = 'High'
- MEDIUM = 'Medium'
- LOW = 'Low'
- class siege_utilities.analytics.RiskSignal[source]
Bases:
objectA named risk signal with type and weight.
- Parameters:
name (str) – Signal identifier.
signal_type (SignalType) – Category of risk this signal measures.
weight (float) – Relative weight (0, 100]. All signals’ weights must sum to 100.
- signal_type: SignalType
- __init__(name, signal_type, weight)
- Parameters:
name (str)
signal_type (SignalType)
weight (float)
- Return type:
None
- class siege_utilities.analytics.RiskTierConfig[source]
Bases:
objectThresholds for risk tier classification.
Scores >= critical_min are Critical, >= high_min are High, etc.
- class siege_utilities.analytics.RiskAssessment[source]
Bases:
objectResult of a risk assessment.
- signal_types: dict[str, SignalType]
- class siege_utilities.analytics.RiskAnalyzer[source]
Bases:
objectMulti-signal risk stacking engine.
- Parameters:
signals (list[RiskSignal]) – Risk signals to evaluate. Weights must sum to 100.
tier_config (RiskTierConfig, optional) – Tier classification thresholds.
interventions (dict[RiskTier, str], optional) – Mapping from risk tier to recommended intervention.
- Raises:
ValueError – If signals is empty or weights don’t sum to 100.
- __init__(signals, tier_config=None, interventions=None)[source]
- Parameters:
signals (list[RiskSignal])
tier_config (RiskTierConfig | None)
- Return type:
None
- assess(entity_id, scores)[source]
Assess risk for an entity.
- Parameters:
- Return type:
- Raises:
ValueError – If scores dict is missing signals or values are out of range.
- class siege_utilities.analytics.DimensionScore[source]
Bases:
objectA score for one entity on one dimension.
- Parameters:
- class siege_utilities.analytics.ComparisonResult[source]
Bases:
objectResult of comparing entities across dimensions.
- class siege_utilities.analytics.FeatureMatrix[source]
Bases:
objectBoolean feature support matrix with coverage scoring.
- class siege_utilities.analytics.ComparativeAnalyzer[source]
Bases:
objectN-entity x M-dimension comparison engine.
- Parameters:
- Raises:
ValueError – If dimensions is empty or weights don’t match dimensions.
- compare(entity_scores)[source]
Compare entities across all dimensions.
- Parameters:
entity_scores (dict[str, dict[str, DimensionScore]]) – Outer key = entity name, inner key = dimension name.
- Return type:
- Raises:
ValueError – If any entity is missing a dimension score.
- static build_feature_matrix(entity_features)[source]
Build a boolean feature support matrix with coverage scores.
- class siege_utilities.analytics.ForecastAccuracy[source]
Bases:
objectResult of a forecast accuracy calculation.
- class siege_utilities.analytics.CategoryBreakdown[source]
Bases:
objectPer-category forecast accuracy.
- accuracy: ForecastAccuracy
- __init__(category, accuracy)
- Parameters:
category (str)
accuracy (ForecastAccuracy)
- Return type:
None
- class siege_utilities.analytics.ForecastReport[source]
Bases:
objectComplete forecast accuracy report.
- overall: ForecastAccuracy
- by_category: list[CategoryBreakdown]
- trend: list[TrendPoint]
- __init__(overall, by_category, trend, trend_direction)
- Parameters:
overall (ForecastAccuracy)
by_category (list[CategoryBreakdown])
trend (list[TrendPoint])
trend_direction (str | None)
- Return type:
None
- class siege_utilities.analytics.ForecastAnalyzer[source]
Bases:
objectForecast accuracy engine.
- Parameters:
- Raises:
ValueError – If inputs have mismatched lengths or are empty.
- class siege_utilities.analytics.PipelineItem[source]
Bases:
objectA single item in the pipeline.
- Parameters:
- class siege_utilities.analytics.StageMetrics[source]
Bases:
objectMetrics for a single pipeline stage.
- class siege_utilities.analytics.ConversionRate[source]
Bases:
objectConversion rate between consecutive stages.
- class siege_utilities.analytics.ConcentrationAlert[source]
Bases:
objectAlert when a single item dominates the pipeline.
- class siege_utilities.analytics.PipelineReport[source]
Bases:
objectComplete pipeline health report.
- stage_metrics: list[StageMetrics]
- conversions: list[ConversionRate]
- velocity: list[VelocityMetrics]
- concentration_alerts: list[ConcentrationAlert]
- __init__(coverage_ratio, stage_metrics, conversions, velocity, concentration_alerts, total_value, target)
- Parameters:
coverage_ratio (float | None)
stage_metrics (list[StageMetrics])
conversions (list[ConversionRate])
velocity (list[VelocityMetrics])
concentration_alerts (list[ConcentrationAlert])
total_value (float)
target (float | None)
- Return type:
None
- class siege_utilities.analytics.PipelineAnalyzer[source]
Bases:
objectPipeline health metrics engine.
- Parameters:
stages (list[str]) – Ordered stage names (first = entry, last = exit).
target (float | None) – Target pipeline value for coverage ratio calculation.
aging_threshold (float) – Multiplier for aging alerts (default 2.0 = alert at 2x average).
concentration_threshold (float) – Maximum single-item share before alert (default 0.40 = 40%).
- Raises:
ValueError – If stages is empty or thresholds are invalid.
- analyze(items, *, as_of=None)[source]
Analyze pipeline health.
- Parameters:
items (list[PipelineItem]) – Items currently in the pipeline.
as_of (datetime, optional) – Reference time for velocity calculation. Defaults to datetime.now().
- Return type:
- Raises:
ValueError – If items is empty or contains unknown stages.
- class siege_utilities.analytics.MappingRule[source]
Bases:
objectA signal-to-action mapping rule.
- Parameters:
- class siege_utilities.analytics.Outcome[source]
Bases:
objectRecorded outcome for an intervention.
- class siege_utilities.analytics.EffectivenessReport[source]
Bases:
objectEffectiveness statistics for one action.
- class siege_utilities.analytics.InterventionMapper[source]
Bases:
objectRule-based intervention mapping engine.
- Parameters:
rules (list[MappingRule]) – Signal-to-action mapping rules.
default_action (str | None) – Action for unmapped signals. If None, unmapped signals raise.
strict (bool) – If True, unmapped signals raise ValueError. If False, warn and return default_action.
- Raises:
ValueError – If rules is empty or strict=False without default_action.
- __init__(rules, default_action=None, strict=True)[source]
- Parameters:
rules (list[MappingRule])
default_action (str | None)
strict (bool)
- Return type:
None
- map(signal, *, override_action=None, override_reason=None)[source]
Map a signal to an action.
- Parameters:
- Returns:
The recommended action.
- Return type:
- Raises:
ValueError – If signal is unmapped and strict=True, or if override is given without a reason.
- record_outcome(signal, action, success, *, overridden=False, override_reason=None)[source]
Record the outcome of an intervention.
- Parameters:
- Raises:
ValueError – If overridden=True without override_reason.
- Return type:
None
- effectiveness()[source]
Compute effectiveness per action.
- Returns:
One entry per distinct action, sorted by success rate ascending.
- Return type:
- Raises:
ValueError – If no outcomes have been recorded.
Submodules
Google Analytics Integration Module
This module provides comprehensive Google Analytics integration capabilities: - Connect to GA accounts via OAuth2 - Retrieve data from GA4 and Universal Analytics - Save data as Pandas or Spark dataframes - Associate GA accounts with client profiles - Batch data retrieval and processing
- class siege_utilities.analytics.google_analytics.GoogleAnalyticsConnector[source]
Bases:
objectGoogle Analytics connection and data retrieval manager.
- __init__(client_id=None, client_secret=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', auth_method='oauth2', service_account_data=None)[source]
Initialize Google Analytics connector.
- Parameters:
client_id (str) – OAuth2 client ID from Google Cloud Console (for OAuth2)
client_secret (str) – OAuth2 client secret from Google Cloud Console (for OAuth2)
redirect_uri (str) – OAuth2 redirect URI (for OAuth2)
auth_method (str) – Authentication method (“oauth2” or “service_account”)
service_account_data (Dict[str, Any]) – Service account credentials dict (for service account auth)
- authenticate_service_account()[source]
Authenticate with Google Analytics using service account. Based on working implementation from GA project.
- Raises:
ValueError – If no service account data is available.
google.auth.exceptions.* – On authentication failure.
OSError – If temp file operations fail.
- Return type:
None
- get_ga4_data(property_id, start_date, end_date, metrics, dimensions=None, row_limit=100000)[source]
Retrieve data from Google Analytics 4.
- Parameters:
property_id (str) – GA4 property ID
start_date (str) – Start date as
YYYY-MM-DD(also accepts the GA4 relative keywordstoday,yesterday, orNdaysAgo). Interpreted in the property’s configured timezone, not the caller’s local time — set this in the GA admin UI.end_date (str) – End date, same format / TZ rules as
start_date.row_limit (int) – Maximum rows to retrieve
- Returns:
Pandas DataFrame with GA4 data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate or recognized GA4 relative keyword. Previously, malformed dates were forwarded raw to the GA API and returned an opaque 400 hours later in a batch run.- Return type:
- get_ua_data(view_id, start_date, end_date, metrics, dimensions=None, max_results=100000)[source]
Retrieve data from Universal Analytics.
- Parameters:
- Returns:
Pandas DataFrame with UA data
- Raises:
ValueError – If
start_date/end_dateis not a validYYYY-MM-DDdate or recognized GA relative keyword.- Return type:
- save_as_pandas(df, output_path, format='parquet')[source]
Save DataFrame as Pandas format.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to save
output_path (str) – Output file path
format (str) – Output format (parquet, csv, excel, etc.)
- Raises:
ValueError – If the format is unsupported.
OSError – If the file cannot be written.
- Return type:
None
- save_as_spark(df, output_path, spark_session=None)[source]
Save DataFrame as Spark DataFrame and optionally to storage.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to convert
output_path (str) – Output path for Spark DataFrame
spark_session (pyspark.sql.SparkSession | None) – Optional SparkSession (will create if not provided)
- Raises:
ImportError – If PySpark is not available.
OSError – If the output path cannot be written.
- Return type:
None
- siege_utilities.analytics.google_analytics.batch_retrieve_ga_data(client_id, start_date, end_date, metrics, dimensions=None, output_format='pandas', output_directory='output/ga_data')[source]
Batch retrieve GA data for all accounts associated with a client.
- Parameters:
- Returns:
Dictionary with retrieval results
- Return type:
- siege_utilities.analytics.google_analytics.create_ga_account_profile(client_id, ga_property_id, account_type='ga4', credentials_file=None)[source]
Create a Google Analytics account profile linked to a client.
- Parameters:
- Returns:
GA account profile dictionary
- Return type:
- siege_utilities.analytics.google_analytics.create_ga_connector_from_1password(item_title='Google Analytics Service Account - Multi-Client Reporter')[source]
Create a GoogleAnalyticsConnector using service account from 1Password.
Deprecated since version The: auth mechanism (1Password) should not leak into an analytics API surface. Use the explicit two-step path instead:
from siege_utilities.config import get_google_service_account_from_1password from siege_utilities.analytics import create_ga_connector_with_service_account data = get_google_service_account_from_1password(item_title) connector = create_ga_connector_with_service_account(data)
This function will be removed in v4.0.0. See docs/INTENT.md (D1) and ELE-2442 for the rationale.
- Parameters:
item_title (str) – Title of the 1Password item containing service account
- Returns:
GoogleAnalyticsConnector instance or None if failed
- Return type:
GoogleAnalyticsConnector | None
- siege_utilities.analytics.google_analytics.create_ga_connector_with_oauth2(client_id, client_secret, redirect_uri='urn:ietf:wg:oauth:2.0:oob')[source]
Create a GoogleAnalyticsConnector using OAuth2 authentication.
- Parameters:
- Returns:
GoogleAnalyticsConnector instance configured for OAuth2 auth
- Return type:
- siege_utilities.analytics.google_analytics.create_ga_connector_with_service_account(service_account_data=None)[source]
Create a GoogleAnalyticsConnector using service account authentication.
- siege_utilities.analytics.google_analytics.list_ga_accounts_for_client(client_id, config_directory='config')[source]
List all GA accounts associated with a client.
- siege_utilities.analytics.google_analytics.load_ga_account_profile(account_id, config_directory='config')[source]
Load GA account profile from JSON file.
- siege_utilities.analytics.google_analytics.save_ga_account_profile(profile, config_directory='config')[source]
Save GA account profile to JSON file.
Google Docs write service.
Provides functions for creating documents, inserting text, tables, images, and performing batch updates via the Docs API v1.
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient from siege_utilities.analytics.google_docs import (
create_document, insert_text, insert_table,
)
client = GoogleWorkspaceClient.from_service_account() doc_id = create_document(client, “Meeting Notes”) insert_text(client, doc_id, “Agendan”, bold=True) insert_table(client, doc_id, rows=3, cols=2)
- siege_utilities.analytics.google_docs.batch_update(client, document_id, requests)[source]
Execute a batch of Docs API requests.
Delegates to
GoogleWorkspaceClient.batch_update_document().
- siege_utilities.analytics.google_docs.copy_document(client, document_id, title=None)[source]
Copy a document via the Drive API.
- siege_utilities.analytics.google_docs.create_document(client, title, folder_id=None)[source]
Create a new Google Doc and return its document ID.
- siege_utilities.analytics.google_docs.get_document(client, document_id)[source]
Fetch full document metadata and content structure.
- siege_utilities.analytics.google_docs.insert_image(client, document_id, image_uri, index=1, width=None, height=None)[source]
Insert an image from a URL.
The URI must be publicly accessible or a Google Drive file.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_docs.insert_paragraph(client, document_id, text, index=1, heading=None)[source]
Insert a paragraph with optional heading style.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
document_id (str) – Target document ID.
text (str) – Paragraph text (newline is appended automatically).
index (int) – Character index (1-based).
heading (str | None) – Named style (
"HEADING_1"through"HEADING_6","TITLE","SUBTITLE", orNonefor normal text).
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_docs.insert_table(client, document_id, rows, cols, index=1)[source]
Insert an empty table at the given index.
- siege_utilities.analytics.google_docs.insert_text(client, document_id, text, index=1, bold=False, italic=False, font_size=None)[source]
Insert text at a given index in a document.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
document_id (str) – Target document ID.
text (str) – Text to insert.
index (int) – Character index (1-based; 1 = start of document body).
bold (bool) – Apply bold formatting.
italic (bool) – Apply italic formatting.
font_size (int | None) – Font size in points (
None= default).
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_docs.read_document_text(client, document_id)[source]
Read the full plain-text content of a document.
Extracts text from all structural elements (paragraphs, tables, etc.).
- siege_utilities.analytics.google_docs.replace_text(client, document_id, find, replace_with, match_case=True)[source]
Find and replace text throughout a document.
Google Sheets write service.
Provides functions for creating spreadsheets, writing data, managing tabs (sheets), and performing batch updates via the Sheets API v4.
All functions accept a GoogleWorkspaceClient for authentication.
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient from siege_utilities.analytics.google_sheets import (
create_spreadsheet, write_dataframe, append_rows,
)
client = GoogleWorkspaceClient.from_service_account() spreadsheet_id = create_spreadsheet(client, “Q1 Report”) write_dataframe(client, spreadsheet_id, df)
- siege_utilities.analytics.google_sheets.add_sheet(client, spreadsheet_id, title)[source]
Add a new tab/sheet to an existing spreadsheet.
- siege_utilities.analytics.google_sheets.append_rows(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Append rows after existing data in a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_sheets.batch_update(client, spreadsheet_id, requests)[source]
Execute a batch of Sheets API requests.
Delegates to
GoogleWorkspaceClient.batch_update_spreadsheet().
- siege_utilities.analytics.google_sheets.copy_spreadsheet(client, spreadsheet_id, title=None)[source]
Copy an entire spreadsheet via the Drive API.
- siege_utilities.analytics.google_sheets.create_spreadsheet(client, title, sheet_names=None, folder_id=None)[source]
Create a new Google Spreadsheet and return its ID.
- Parameters:
- Returns:
The spreadsheet ID string.
- Return type:
- siege_utilities.analytics.google_sheets.get_spreadsheet_metadata(client, spreadsheet_id)[source]
Fetch spreadsheet metadata (title, sheets, locale, etc.).
- siege_utilities.analytics.google_sheets.read_dataframe(client, spreadsheet_id, range_='Sheet1', has_header=True)[source]
Read a range from a spreadsheet into a pandas DataFrame.
- siege_utilities.analytics.google_sheets.read_values(client, spreadsheet_id, range_)[source]
Read values from a range.
- siege_utilities.analytics.google_sheets.write_dataframe(client, spreadsheet_id, df, sheet_name='Sheet1', include_header=True, start_cell='A1')[source]
Write a pandas DataFrame to a sheet.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_sheets.write_values(client, spreadsheet_id, range_, values, value_input_option='USER_ENTERED')[source]
Write a 2-D list of values to a range.
- Parameters:
- Returns:
The API response dict.
- Return type:
Google Slides write service.
Provides functions for creating presentations, adding slides, inserting text and images, and performing batch updates via the Slides API v1.
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient from siege_utilities.analytics.google_slides import (
create_presentation, add_blank_slide, insert_text,
)
client = GoogleWorkspaceClient.from_service_account() pres_id = create_presentation(client, “Q1 Report”) slide_id = add_blank_slide(client, pres_id) insert_text(client, pres_id, slide_id, “Hello World”)
- siege_utilities.analytics.google_slides.add_blank_slide(client, presentation_id, layout='BLANK', insertion_index=None)[source]
Add a blank slide to a presentation.
- Parameters:
- Returns:
The new slide’s object ID.
- Return type:
- siege_utilities.analytics.google_slides.batch_update(client, presentation_id, requests)[source]
Execute a batch of Slides API requests.
Delegates to
GoogleWorkspaceClient.batch_update_presentation().
- siege_utilities.analytics.google_slides.copy_presentation(client, presentation_id, title=None)[source]
Copy an entire presentation via the Drive API.
- siege_utilities.analytics.google_slides.create_argument_slide(client, presentation_id, argument, slide_index=None)[source]
Add one slide for an Argument.
- Layout is derived from argument.layout:
“full_width” → title / narrative / figure stacked vertically “side_by_side” → title top; narrative left, figure right
- siege_utilities.analytics.google_slides.create_presentation(client, title, folder_id=None)[source]
Create a new Google Slides presentation and return its ID.
- siege_utilities.analytics.google_slides.create_report_from_arguments(client, title, arguments, folder_id=None, theme_presentation_id=None)[source]
Create a complete presentation from a list of Arguments.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
title (str) – Presentation title.
arguments (List) – List of Argument dataclass instances (one slide each).
folder_id (str | None) – Optional Drive folder to place the presentation in.
theme_presentation_id (str | None) – If given, copies this presentation first (preserves master slides / theme).
- Returns:
Presentation ID of the newly created report.
- Return type:
- siege_utilities.analytics.google_slides.create_textbox(client, presentation_id, slide_id, text, left=100, top=100, width=400, height=50)[source]
Create a text box on a slide and populate it with text.
Dimensions are in EMU (English Metric Units). 1 inch = 914400 EMU. For convenience, this function accepts points (1 pt = 12700 EMU) but the values are treated as points internally.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
presentation_id (str) – Target presentation ID.
slide_id (str) – Slide object ID.
text (str) – Text to put in the box.
left (float) – Position in points from top-left.
top (float) – Position in points from top-left.
width (float) – Dimensions in points.
height (float) – Dimensions in points.
- Returns:
The textbox object ID.
- Return type:
- siege_utilities.analytics.google_slides.get_presentation(client, presentation_id)[source]
Fetch full presentation metadata.
- siege_utilities.analytics.google_slides.insert_image(client, presentation_id, slide_id, image_url, left=100, top=100, width=400, height=300)[source]
Insert an image onto a slide from a URL.
The URL must be publicly accessible or a Google Drive file URL.
- Parameters:
client – Authenticated GoogleWorkspaceClient.
presentation_id (str) – Target presentation ID.
slide_id (str) – Slide object ID.
image_url (str) – Public URL of the image.
left (float) – Position in points.
top (float) – Position in points.
width (float) – Size in points.
height (float) – Size in points.
- Returns:
The image object ID.
- Return type:
- siege_utilities.analytics.google_slides.insert_text(client, presentation_id, object_id, text, insertion_index=0)[source]
Insert text into an existing shape or text box.
- Parameters:
- Returns:
The API response dict.
- Return type:
- siege_utilities.analytics.google_slides.upload_figure_to_drive(client, figure, filename, folder_id=None)[source]
Save a matplotlib Figure to a temp PNG, upload to Google Drive, return URL.
- Parameters:
- Returns:
Public URL string for the uploaded image (suitable for insert_image).
- Return type:
Google Workspace base client for Docs, Sheets, and Slides write APIs.
Provides shared authentication and service-building logic reused by the Sheets, Slides, and Docs service modules.
Authentication follows the same patterns as GoogleAnalyticsConnector: - OAuth2 (interactive flow with token file) - Service Account (from 1Password or file) - Explicit Credentials object
- Usage:
from siege_utilities.analytics.google_workspace import GoogleWorkspaceClient
# OAuth2 with token file client = GoogleWorkspaceClient.from_oauth(
client_id=”…”, client_secret=”…”, token_file=”token.json”,
)
# Service Account from 1Password client = GoogleWorkspaceClient.from_service_account()
# Get a specific API service sheets = client.sheets_service() docs = client.docs_service() slides = client.slides_service()
- class siege_utilities.analytics.google_workspace.GoogleWorkspaceClient[source]
Bases:
objectAuthenticated client that builds Google API service objects.
Do not instantiate directly — use the
from_oauth()orfrom_service_account()class methods.- classmethod from_oauth(client_id, client_secret, token_file=None, redirect_uri='urn:ietf:wg:oauth:2.0:oob', scopes=None)[source]
Authenticate via OAuth2 interactive flow.
If token_file exists and contains a valid/refreshable token, no browser interaction is needed.
- classmethod from_1password(item_title='Google OAuth Client - siege_utilities', vault=None, account='TLTQ3ANAABGCNEK7KIAOTDNK2Q', token_file=None, scopes=None)[source]
Authenticate using credentials stored in a 1Password Document item.
Auto-detects whether the JSON document is an OAuth client secret (has
"installed"or"web"key) or a service account key (has"type": "service_account"), and routes to the appropriate auth flow.For OAuth: runs the installed-app flow (browser required on first use; cached token reused afterward).
For service account: authenticates server-to-server with no browser interaction.
- Parameters:
item_title (str) – Title of the 1Password Document item.
vault (str | None) – 1Password vault name.
account (str | None) – 1Password account shorthand or UUID.
token_file (str | Path | None) – Path to cache the OAuth token (ignored for service accounts). Default:
~/.siege/tokens/workspace_token.json.scopes (List[str] | None) – OAuth scopes (defaults to
WORKSPACE_SCOPES).
- Returns:
Authenticated GoogleWorkspaceClient.
- Return type:
- classmethod from_service_account(service_account_data=None, service_account_file=None, scopes=None)[source]
Authenticate via service account credentials.
Provide service_account_data (dict) or service_account_file (path). If neither is given, attempts to fetch from 1Password via
CredentialManager.get_google_service_account_from_1password().
- classmethod from_credentials(credentials)[source]
Wrap an already-authenticated
google.auth.credentials.Credentials.- Return type:
- classmethod from_account(account, person=None, scopes=None)[source]
Build a client from a
GoogleAccount.For OAuth accounts the method looks for a cached token_file first, then resolves oauth_integration_name from person to get client_id/client_secret for the OAuth flow.
For service accounts the method resolves service_account_ref via
CredentialManager(1Password) or treats it as a file path.- Parameters:
account (GoogleAccount)
person (Optional[Person])
scopes (Optional[List[str]])
- Return type:
- classmethod from_registry(registry, google_account_id=None, person=None, scopes=None)[source]
Build a client from a registry’s default or specified account.
If google_account_id is given, looks it up directly; otherwise uses the registry’s default account.
- Parameters:
registry (GoogleAccountRegistry)
google_account_id (Optional[str])
person (Optional[Person])
scopes (Optional[List[str]])
- Return type:
- property credentials
- batch_update_presentation(presentation_id, requests)[source]
Execute a batch of Slides API requests.
- static spreadsheet_url(spreadsheet_id)[source]
Return the live Google Sheets URL for a spreadsheet ID.
- static presentation_url(presentation_id)[source]
Return the live Google Slides URL for a presentation ID.
- static file_url(file_id, mime_type=None)[source]
Return the live Google Drive URL for a file ID.
If mime_type is provided, returns the appropriate editor URL. Otherwise returns the generic Drive file URL.
- copy_file(file_id, title=None)[source]
Copy a Drive file (spreadsheet, doc, presentation) and return the new ID.
Share a Drive file with a user.
Snowflake data warehouse connector for Siege Utilities.
Provides seamless integration with Snowflake for data warehousing and analytics.
- class siege_utilities.analytics.snowflake_connector.SnowflakeConnector[source]
Bases:
objectSnowflake data warehouse connector with advanced features.
- __init__(account, user, password=None, warehouse=None, database=None, schema=None, role=None, config_file=None)[source]
Initialize Snowflake connector.
- Parameters:
account (str) – Snowflake account identifier
user (str) – Snowflake username
password (str | None) – Snowflake password (optional if using config file)
warehouse (str | None) – Default warehouse to use
database (str | None) – Default database to use
schema (str | None) – Default schema to use
role (str | None) – Default role to use
config_file (str | Path | None) – Path to configuration file
- connect()[source]
Establish connection to Snowflake.
- Raises:
ConnectionError – If the connection attempt fails.
- Return type:
None
- execute_query(query, params=None)[source]
Execute a SQL query.
- Parameters:
- Returns:
Query results as list of tuples
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If query execution fails.
- Return type:
- execute_ddl(ddl_statement)[source]
Execute DDL statements (CREATE, ALTER, DROP, etc.).
- Parameters:
ddl_statement (str) – DDL SQL statement
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If DDL execution fails.
- Return type:
None
- upload_dataframe(df, table_name, database=None, schema=None, auto_create_table=True, overwrite=False)[source]
Upload pandas DataFrame to Snowflake table.
- Parameters:
df (pd.DataFrame) – Pandas DataFrame to upload
table_name (str) – Target table name
database (str | None) – Target database (uses default if not specified)
schema (str | None) – Target schema (uses default if not specified)
auto_create_table (bool) – Whether to automatically create table if it doesn’t exist
overwrite (bool) – Whether to overwrite existing table
- Raises:
ImportError – If pandas is not available.
ConnectionError – If not connected and connection fails.
RuntimeError – If the upload fails.
- Return type:
None
- download_dataframe(query, params=None)[source]
Download data from Snowflake as pandas DataFrame.
- Parameters:
- Returns:
Pandas DataFrame with query results
- Raises:
ImportError – If pandas is not available.
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
pd.DataFrame
- get_table_info(table_name, database=None, schema=None)[source]
Get information about a Snowflake table.
- Parameters:
- Returns:
Dictionary with table information
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- list_tables(database=None, schema=None)[source]
List all tables in a database/schema.
- Parameters:
- Returns:
List of table names
- Raises:
ConnectionError – If not connected and connection fails.
RuntimeError – If the query fails.
- Return type:
- siege_utilities.analytics.snowflake_connector.get_snowflake_connector(config_file=None)[source]
Get Snowflake connector instance from configuration.
- Parameters:
- Return type:
- siege_utilities.analytics.snowflake_connector.upload_to_snowflake(df, table_name, config_file=None, **kwargs)[source]
Convenience function to upload DataFrame to Snowflake.
- siege_utilities.analytics.snowflake_connector.download_from_snowflake(query, config_file=None, **kwargs)[source]
Convenience function to download DataFrame from Snowflake.
- siege_utilities.analytics.snowflake_connector.execute_snowflake_query(query, config_file=None, **kwargs)[source]
Convenience function to execute Snowflake query.
Data.world connector for Siege Utilities.
Provides seamless integration with data.world for data discovery, access, and collaboration.
- class siege_utilities.analytics.datadotworld_connector.DataDotWorldConnector[source]
Bases:
objectData.world connector with advanced data discovery and access features.
- search_datasets(query, limit=50, owner=None, tags=None)[source]
Search for datasets on data.world.
- Parameters:
- Returns:
List of dataset information dictionaries
- Raises:
ConnectionError – If the API request fails.
- Return type:
- get_dataset_info(dataset_id)[source]
Get detailed information about a specific dataset.
- Parameters:
dataset_id (str) – Dataset identifier (format: owner/dataset)
- Returns:
Dictionary with detailed dataset information
- Raises:
ConnectionError – If the API request fails.
- Return type:
- download_file(dataset_id, file_name, output_path=None)[source]
Download a specific file from a dataset.
- Parameters:
- Returns:
Path to the downloaded file
- Raises:
ConnectionError – If the download fails.
OSError – If writing to disk fails.
- Return type:
- load_dataset_as_dataframe(dataset_id, file_name=None)[source]
Load a dataset file directly as a pandas DataFrame.
- Parameters:
- Returns:
Pandas DataFrame with the data
- Raises:
ImportError – If pandas is not available.
ConnectionError – If the API request fails.
- Return type:
pd.DataFrame
- query_dataset(dataset_id, query, query_type='sql')[source]
Execute a query against a dataset.
- Parameters:
- Returns:
Pandas DataFrame with query results
- Raises:
ImportError – If pandas is not available.
ValueError – If query_type is invalid.
ConnectionError – If the query fails.
- Return type:
pd.DataFrame
- get_dataset_schema(dataset_id)[source]
Get the schema information for a dataset.
- Parameters:
dataset_id (str) – Dataset identifier (format: owner/dataset)
- Returns:
Dictionary with schema information
- Raises:
ConnectionError – If the API request fails.
- Return type:
- create_dataset(owner, dataset_id, title, description, tags=None, license='Public Domain', visibility='OPEN')[source]
Create a new dataset on data.world.
- Parameters:
- Returns:
owner/dataset_id)
- Return type:
Created dataset ID (format
- Raises:
ValueError – If API token is not configured.
ConnectionError – If the API request fails.
- upload_file(dataset_id, file_path, file_name=None)[source]
Upload a file to an existing dataset.
- Parameters:
- Raises:
ValueError – If API token is not configured.
FileNotFoundError – If the local file does not exist.
ConnectionError – If the upload fails.
- Return type:
None
- update_dataset(dataset_id, title=None, description=None, tags=None, license=None)[source]
Update an existing dataset.
- Parameters:
- Raises:
ValueError – If API token is not configured or no updates provided.
ConnectionError – If the API request fails.
- Return type:
None
- delete_dataset(dataset_id)[source]
Delete a dataset.
- Parameters:
dataset_id (str) – Dataset identifier (format: owner/dataset)
- Raises:
ValueError – If API token is not configured.
ConnectionError – If the API request fails.
- Return type:
None
- siege_utilities.analytics.datadotworld_connector.get_datadotworld_connector(config_file=None)[source]
Get Data.world connector instance from configuration.
- Parameters:
- Return type:
- siege_utilities.analytics.datadotworld_connector.list_datasets(limit=50, config_file=None, **kwargs)[source]
Standalone function to list available datasets.
- siege_utilities.analytics.datadotworld_connector.search_datasets(query, config_file=None, **kwargs)[source]
Standalone function to search for datasets.
- siege_utilities.analytics.datadotworld_connector.search_datadotworld_datasets(query, config_file=None, **kwargs)[source]
Convenience function to search for datasets.
- siege_utilities.analytics.datadotworld_connector.load_datadotworld_dataset(dataset_id, config_file=None, **kwargs)[source]
Convenience function to load a dataset as DataFrame.
- siege_utilities.analytics.datadotworld_connector.query_datadotworld_dataset(dataset_id, query, config_file=None, **kwargs)[source]
Convenience function to query a dataset.
- Parameters:
- Return type:
pd.DataFrame
Facebook Business Integration Module
This module provides comprehensive Facebook Business integration capabilities: - Connect to Facebook Business accounts via OAuth2 - Retrieve data from Facebook Ads, Pages, and Business Manager - Save data as Pandas or Spark dataframes - Associate Facebook accounts with client profiles - Batch data retrieval and processing
- class siege_utilities.analytics.facebook_business.FacebookBusinessConnector[source]
Bases:
objectFacebook Business connection and data retrieval manager.
- __init__(access_token, app_id=None, app_secret=None)[source]
Initialize Facebook Business connector.
- get_ad_insights(ad_account_id, start_date, end_date, fields=None, breakdowns=None)[source]
Retrieve ad insights from Facebook Ads.
- Parameters:
- Returns:
Pandas DataFrame with ad insights
- Return type:
- get_page_insights(page_id, start_date, end_date, metrics=None)[source]
Retrieve page insights from Facebook Pages.
- get_business_insights(business_id, start_date, end_date, metrics=None)[source]
Retrieve business insights from Facebook Business Manager.
- save_as_pandas(df, output_path, format='parquet')[source]
Save DataFrame as Pandas format.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to save
output_path (str) – Output file path
format (str) – Output format (parquet, csv, excel, etc.)
- Raises:
ValueError – If the format is unsupported.
OSError – If the file cannot be written.
- Return type:
None
- save_as_spark(df, output_path, spark_session=None)[source]
Save DataFrame as Spark DataFrame and optionally to storage.
- Parameters:
df (pandas.DataFrame) – Pandas DataFrame to convert
output_path (str) – Output path for Spark DataFrame
spark_session (pyspark.sql.SparkSession | None) – Optional SparkSession (will create if not provided)
- Raises:
ImportError – If PySpark is not available.
OSError – If the output path cannot be written.
- Return type:
None
- siege_utilities.analytics.facebook_business.batch_retrieve_facebook_data(client_id, start_date, end_date, data_types=None, output_format='pandas', output_directory='output/facebook_data')[source]
Batch retrieve Facebook data for all accounts associated with a client.
- Parameters:
client_id (str) – Client ID to retrieve data for
start_date (str) – Start date for data retrieval
end_date (str) – End date for data retrieval
data_types (List[str]) – Types of data to retrieve (ads, pages, business)
output_format (str) – Output format (pandas, spark, or both)
output_directory (str) – Directory to save output files
- Returns:
Dictionary with retrieval results
- Return type:
- siege_utilities.analytics.facebook_business.create_facebook_account_profile(client_id, fb_account_id, account_type='ad_account', access_token=None)[source]
Create a Facebook account profile linked to a client.
- siege_utilities.analytics.facebook_business.list_facebook_accounts_for_client(client_id, config_directory='config')[source]
List all Facebook accounts associated with a client.
- siege_utilities.analytics.facebook_business.load_facebook_account_profile(account_id, config_directory='config')[source]
Load Facebook account profile from JSON file.
- siege_utilities.analytics.facebook_business.save_facebook_account_profile(profile, config_directory='config')[source]
Save Facebook account profile to JSON file.
Vista Social (https://vistasocial.com) is a social-media management
platform that aggregates engagement, posts, and audience metrics across
Facebook, Instagram, LinkedIn, X, TikTok, YouTube, and other channels.
This module provides a thin connector that auths against the Vista
Social REST API and exposes a few common reporting calls in the same
shape as the other connectors in this package
(FacebookBusinessConnector, DataDotWorldConnector).
Why “thin”
Vista Social’s REST API documentation requires an authenticated account; the public website does not enumerate endpoint URLs. This connector ships with:
The base
_requesthelper that handles auth headers, retries, and error normalization – fully tested.Concrete
account,profiles,posts,analyticsmethods that wrap likely endpoint paths (/v1/accounts,/v1/profiles,/v1/posts,/v1/analytics). These are informed by the publicly visible URL structure on the Vista Social web app and are documented as “best-known” rather than spec-confirmed.
When you obtain real Vista Social API documentation (request access at
api@vistasocial.com), verify the endpoints against the docs and adjust
this module. The _request plumbing is correct regardless of which
endpoints exist.
The intended consumer is siege_utilities.reporting, which will
embed the returned analytics into a combined GA4 + social-media PDF
report (the second half of #306). That work lives in a follow-up PR.
- exception siege_utilities.analytics.vista_social.VistaSocialError[source]
Bases:
RuntimeErrorBase class for all Vista Social connector failures.
- exception siege_utilities.analytics.vista_social.VistaSocialAuthError[source]
Bases:
VistaSocialErrorAuthentication failed (401/403). Token is missing, wrong, or expired.
- exception siege_utilities.analytics.vista_social.VistaSocialRateLimitError[source]
Bases:
VistaSocialErrorAPI rate limit hit (429). Caller can back off and retry later.
- class siege_utilities.analytics.vista_social.VistaSocialResponse[source]
Bases:
objectStructured response from a Vista Social API call.
The connector returns this rather than the raw
requests.Responseso consumers don’t need to know about the underlying HTTP library.- data
Parsed JSON body. Empty dict if the response had no body (e.g. 204 No Content) or wasn’t JSON-decodable.
- Type:
Mapping[str, Any]
- headers
Response headers – useful for inspecting rate-limit counters (
X-RateLimit-Remaining,Retry-After).
- class siege_utilities.analytics.vista_social.VistaSocialConnector[source]
Bases:
objectBearer-token authenticated client for the Vista Social REST API.
Usage:
from siege_utilities.analytics.vista_social import VistaSocialConnector client = VistaSocialConnector(api_token=os.environ["VISTA_SOCIAL_TOKEN"]) accounts = client.list_accounts() analytics = client.get_account_analytics( account_id=accounts[0]["id"], start_date="2026-01-01", end_date="2026-01-31", )
All methods raise:
VistaSocialAuthErroron 401/403,VistaSocialRateLimitErroron 429,VistaSocialErroron other non-2xx responses or network failures.
The connector follows the project’s failure-mode discipline (see
docs/FAILURE_MODES.md): no method silently returns an empty dict on failure. Every failure mode is either a typed exception or a None return with a documented meaning.- __init__(api_token, *, base_url='https://api.vistasocial.com', timeout=30, retry_attempts=3, retry_backoff=1.5)[source]
- list_accounts()[source]
Return the list of accounts this token can access.
Wraps
GET /v1/accounts. Returns the decodeddatalist from the response, or an empty list if the API returned nothing. Raises on auth / rate-limit / network failures (no silent empty-on-error).
- list_profiles(account_id)[source]
Return social profiles connected to account_id.
Wraps
GET /v1/accounts/{account_id}/profiles.
- get_account_analytics(account_id, *, start_date, end_date, metrics=None)[source]
Fetch aggregate analytics for account_id over a date range.
- Parameters:
account_id (str) – Vista Social account id (string, opaque).
start_date (str) – ISO-8601 date (
YYYY-MM-DD), inclusive.end_date (str) – ISO-8601 date, inclusive.
metrics (List[str] | None) – Optional list of metric names (e.g.
["impressions", "engagements", "reach", "follower_count"]). If omitted, Vista Social returns its default metric set.
- Return type:
Wraps
GET /v1/accounts/{account_id}/analytics. Returns the decoded JSON body verbatim – Vista Social returns a structured document with metric breakdowns; consumers can extract what they need.
- list_posts(account_id, *, start_date=None, end_date=None, limit=100)[source]
Return posts for account_id with optional date filtering.
Wraps
GET /v1/accounts/{account_id}/posts. Pagination is intentionally simple here – consumers that need full enumeration should call repeatedly with adjusted date ranges or extend this connector with cursor handling once the real API spec is known.