Schema Provider Module

Overview

The schema_provider module extracts database schema (field names) from CSV and Google Sheets sources.

Used by the filter editor to:

Detect database type (CSV vs Google Sheets)
Extract field names from headers
Provide field suggestions and validation

Database schema extraction for filter validation.

Extracts field names (headers) from CSV and Google Sheets databases. Used by filter editor to provide field name suggestions and validate that filter fields exist in the subscriber database.

Classes:: DatabaseSchemaProvider: Detects database type and extracts schema

class schema_provider.DatabaseSchemaProvider[source]

Bases: object

Extract field names (schema) from database sources.

static from_csv(csv_path: str) → list[str][source]

Extract field names from CSV file headers.

Parameters:: csv_path – Path to CSV file
Returns:: List of field names from first row (headers)

static from_google_sheets(service: Any, spreadsheet_id: str, sheet_name: str | None = None) → list[str][source]

Extract field names from Google Sheets first row.

Parameters:

service – Google Sheets service object (from gspread or google-api-python-client)
spreadsheet_id – Spreadsheet ID (unused in gspread, service already bound)
sheet_name – Sheet name (if None, uses first sheet)

Returns:

List of field names from first row

static detect_and_extract(database_path: str, sheet_name: str | None = None, gsheet_service: Any = None) → list[str][source]

Detect database type and extract schema.

Parameters:

database_path – Path to CSV file or Google Sheets URL/ID
sheet_name – Sheet name for Google Sheets
gsheet_service – Google Sheets service object

Returns:

List of field names

Classes

DatabaseSchemaProvider

class schema_provider.DatabaseSchemaProvider[source]

Bases: object

Extract field names (schema) from database sources.

static from_csv(csv_path: str) → list[str][source]

Extract field names from CSV file headers.

Parameters:: csv_path – Path to CSV file
Returns:: List of field names from first row (headers)

static from_google_sheets(service: Any, spreadsheet_id: str, sheet_name: str | None = None) → list[str][source]

Extract field names from Google Sheets first row.

Parameters:

service – Google Sheets service object (from gspread or google-api-python-client)
spreadsheet_id – Spreadsheet ID (unused in gspread, service already bound)
sheet_name – Sheet name (if None, uses first sheet)

Returns:

List of field names from first row

static detect_and_extract(database_path: str, sheet_name: str | None = None, gsheet_service: Any = None) → list[str][source]

Detect database type and extract schema.

Parameters:

database_path – Path to CSV file or Google Sheets URL/ID
sheet_name – Sheet name for Google Sheets
gsheet_service – Google Sheets service object

Returns:

List of field names

Usage Examples

Extract schema from CSV file

Extract field names from a local CSV database:

from schema_provider import DatabaseSchemaProvider

schema = DatabaseSchemaProvider.from_csv('subscribers.csv')
print(schema)
# Output: ['id', 'first_name', 'last_name', 'email', 'status', ...]

Extract schema from Google Sheets

Extract field names from Google Sheets (requires gspread service object):

from schema_provider import DatabaseSchemaProvider
import gspread

# Assuming you have a gspread service object
service = gspread.authorize(creds)
spreadsheet = service.open_by_key('sheet_id')

schema = DatabaseSchemaProvider.from_google_sheets(
    spreadsheet,
    spreadsheet_id='sheet_id',
    sheet_name='Members'
)

Auto-detect database type

Automatically detect CSV vs Google Sheets and extract schema:

from schema_provider import DatabaseSchemaProvider

# CSV file
schema = DatabaseSchemaProvider.detect_and_extract('subscribers.csv')

# Google Sheets (requires gsheet_service)
schema = DatabaseSchemaProvider.detect_and_extract(
    'https://docs.google.com/spreadsheets/d/...',
    gsheet_service=spreadsheet_obj
)