Skip to content

xo-core Package

xo-core v1.6.0 -- The foundation package of the XO-Data platform. Provides core utilities, data extractors, and managers for common data engineering operations.

Purpose

xo-core serves as the shared utility layer for all XO-Data packages and applications. It contains:

  • Extractors: API data source connectors (Gladly, Sprout Social, Medallia/Stella Connect). Gmail, Google Sheets/Drive, and S3 are accessed via the service managers below (wrapped by foundry extract tasks), not via extractor classes.
  • Managers: Service wrappers (Snowflake, S3, Google Services)
  • Utilities: DataFrame operations, file handling, logging

Installation

# Install xo-core package
uv sync --package xo-core

# Add as dependency to another package
uv add --package xo-foundry xo-core

Package Structure

packages/xo-core/src/xo_core/
├── extractors/               # API data source connectors
│   ├── gladly_extractor.py   # GladlyAPIClient
│   ├── sprout_extractor.py   # SproutAPIClient
│   └── medallia_extractor.py # MedalliaAPIClient (Stella Connect CSAT)
│                             # Gmail / GSheets / Drive / S3 use the
│                             # managers below, not extractor classes
│
├── google_services/          # Google API integrations
│   ├── auth.py               # OAuth2 authentication
│   ├── gmail_manager.py      # Gmail operations
│   ├── gdrive_manager.py     # Google Drive operations
│   └── gsheets_manager.py    # Google Sheets operations (gspread)
│
├── snowflake_manager.py      # Snowflake operations (SnowflakeManager)
├── s3_manager.py             # S3 operations (S3Manager)
├── df_utils.py               # DataFrame cleaning / standardization
├── file_utils.py             # File handling
├── logger.py                 # Logging configuration
└── __init__.py

Key Components

Extractors

Data source connectors that extract data and upload to S3.

No pandas in extraction

xo-foundry extraction tasks (which wrap these extractors) use native Python csv.DictWriter -- never pandas. Pandas corrupts data via automatic type inference (IDs become floats, leading zeros stripped). Let Snowflake handle type conversion.

GladlyAPIClient

Extract reports from the Gladly API.

from xo_core.extractors.gladly_extractor import (
    GladlyAPIClient,
    extract_gladly_reports,
)

client = GladlyAPIClient(
    base_url="https://your-org.gladly.com",
    email="user@example.com",
    token="your-token",
)

# Methods: get_report(), get_report_by_month(), get_agents()
report_data = client.get_report(...)

SproutAPIClient

Extract data from the Sprout Social API.

from xo_core.extractors.sprout_extractor import SproutAPIClient

client = SproutAPIClient(
    api_key="your-api-key",
    client_id="2105997",
)

# Methods: get_messages(), get_all_messages(), get_cases(), get_profile_analytics()
messages = client.get_messages(...)

MedalliaAPIClient

Extract CSAT survey responses from the Medallia (Stella Connect) API.

from xo_core.extractors.medallia_extractor import MedalliaAPIClient

with MedalliaAPIClient(
    api_key="your-api-key",
    token="your-token"
) as client:
    responses = client.get_csat_responses(
        responded_at_gte="2026-01-15",
        responded_at_lte="2026-01-15"
    )

Authentication: Header-based (x-api-key + Authorization token). Retry logic: 3 attempts with exponential backoff on 429 rate-limit responses. Returns: list[dict[str, Any]] — raw API response records, one per CSAT survey response.

Gmail, Google Sheets & S3 — use managers, not extractors

There are no GmailExtractor, GSheetsExtractor, or S3Extractor classes. These sources are accessed through service managers — GmailManager and GSheetsManager (and GDriveManager) under google_services/, and S3Manager — which the foundry extract tasks (extract_gmail_data, extract_gsheet_data, etc.) wrap. Foundry exposes these as gmail, gsheet, and s3 source types.

Managers

Service wrappers that provide high-level operations for external services.

SnowflakeManager

Modern Snowflake connector with DataFrame operations and deduplication.

from xo_core.snowflake_manager import SnowflakeManager

manager = SnowflakeManager()

# Upload DataFrame with deduplication
prepped_df = manager.prep_dataframe_for_table(
    df=raw_df,
    table_name="GLADLY_CONTACT_TIMESTAMPS",
    filter_existing=True
)

manager.upload_dataframe(
    prepped_df,
    table_name="GLADLY_CONTACT_TIMESTAMPS",
    database="WBP_DB",
    schema="BRONZE"
)

# Execute query
results = manager.execute_query("SELECT COUNT(*) FROM TABLE")

# Download table to DataFrame
df = manager.download_table("DATABASE.SCHEMA.TABLE")

Key Features:

  • Automatic deduplication (single-field or multi-field)
  • Type-safe DataFrame uploads
  • Connection pooling
  • Prepared statement support

S3Manager

S3 operations with DataFrame support.

from xo_core.s3_manager import S3Manager

manager = S3Manager()

# Upload DataFrame
manager.upload_dataframe(
    df,
    bucket="xo-ingest",
    key="client/report/2026-01-15/data.csv"
)

# Download to DataFrame
df = manager.download_to_dataframe(
    bucket="xo-ingest",
    key="client/report/2026-01-15/data.csv"
)

# List files
files = manager.list_files(bucket="xo-ingest", prefix="client/")

Google Services

GSheetsManager

Google Sheets operations via gspread library.

from xo_core.google_services.gsheets_manager import GSheetsManager

gsheets = GSheetsManager(credentials_path="credentials.json")

# Read spreadsheet
data = gsheets.read_sheet(spreadsheet_id="abc123", sheet_name="Data")

GmailManager

Gmail API operations.

from xo_core.google_services.gmail_manager import GmailManager

gmail = GmailManager(credentials_path="credentials.json")

# List messages
messages = gmail.list_messages(label="Reports", after_date="2026-01-15")

# Get attachments
attachments = gmail.get_attachments(message_id="abc123")

Utilities

df_utils

DataFrame cleaning and standardization utilities.

from xo_core.df_utils import (
    clean_column_names,
    clean_dataframe,
    datetime_columns_handler,
    convert_to_nullable_int_columns,
    generate_record_key
)

# Clean column names (UPPERCASE, replace special chars)
df.columns = clean_column_names(df.columns)

# Comprehensive DataFrame cleaning
df = clean_dataframe(
    df,
    uppercase_columns=True,
    datetime_columns=['CREATED_AT'],
    auto_detect_int_columns=True,
    replace_empty_strings=True
)

Key Functions:

Function Purpose
clean_column_names() Standardize column names (UPPERCASE, _)
clean_dataframe() All-in-one DataFrame cleaning
datetime_columns_handler() Convert to datetime with timezone
convert_to_nullable_int_columns() Convert to Int64 (allows nulls)
replace_empty_strings_with_na() Replace "" with NA
drop_empty_columns() Remove columns with all nulls
generate_record_key() Generate unique keys for deduplication

logger

Consistent logging configuration.

from xo_core.logger import get_module_logger

logger = get_module_logger(__name__)

logger.info("Processing started")
logger.warning("Missing optional field")
logger.error("Failed to connect", exc_info=True)

Type Safety

xo-core is fully typed and must pass ty with zero errors:

uv run ty check packages/xo-core

from typing import Any
import pandas as pd

def process_data(items: list[dict[str, str]]) -> pd.DataFrame:
    """Process data with proper typing."""
    ...

Testing

# Run xo-core tests
uv run pytest packages/xo-core/tests/

# Run specific test file
uv run pytest packages/xo-core/tests/test_df_utils.py

# Run with coverage
uv run pytest --cov=xo_core packages/xo-core/tests/

Dependencies

Key dependencies:

  • pandas>=2.0 -- DataFrame operations
  • snowflake-connector-python -- Snowflake connectivity
  • boto3 -- AWS S3 operations
  • google-auth -- Google API authentication
  • gspread -- Google Sheets API
  • requests -- HTTP client for APIs
  • tqdm -- Progress bars

Next Steps

Package Location: packages/xo-core/ Version: 1.6.0 Dependencies: See packages/xo-core/pyproject.toml