Create data/schema.py

data/schema.py

@@ -0,0 +1,259 @@
+"""
+Pharmaceutical Analytics Database Schema
+
+This module defines the SQLite database schema for the pharmaceutical analytics demo.
+It includes tables for sales data, products, regions, territories, prescribers,
+pharmacies, marketing campaigns, and other relevant entities.
+"""
+
+import sqlite3
+from typing import Dict, List, Any
+
+# SQLite database schema definitions
+SCHEMA_DEFINITIONS = {
+    "regions": """
+        CREATE TABLE IF NOT EXISTS regions (
+            region_id TEXT PRIMARY KEY,
+            region_name TEXT NOT NULL,
+            country TEXT NOT NULL,
+            division TEXT NOT NULL,
+            population INTEGER NOT NULL
+        )
+    """,
+    
+    "territories": """
+        CREATE TABLE IF NOT EXISTS territories (
+            territory_id TEXT PRIMARY KEY,
+            territory_name TEXT NOT NULL,
+            region_id TEXT NOT NULL,
+            sales_rep_id TEXT NOT NULL,
+            FOREIGN KEY (region_id) REFERENCES regions (region_id)
+        )
+    """,
+    
+    "products": """
+        CREATE TABLE IF NOT EXISTS products (
+            product_id TEXT PRIMARY KEY,
+            product_name TEXT NOT NULL,
+            therapeutic_area TEXT NOT NULL,
+            molecule TEXT NOT NULL,
+            launch_date DATE NOT NULL,
+            status TEXT NOT NULL,
+            list_price REAL NOT NULL
+        )
+    """,
+    
+    "competitor_products": """
+        CREATE TABLE IF NOT EXISTS competitor_products (
+            competitor_product_id TEXT PRIMARY KEY,
+            product_name TEXT NOT NULL,
+            manufacturer TEXT NOT NULL,
+            therapeutic_area TEXT NOT NULL,
+            molecule TEXT NOT NULL,
+            launch_date DATE NOT NULL,
+            list_price REAL NOT NULL,
+            competing_with_product_id TEXT NOT NULL,
+            FOREIGN KEY (competing_with_product_id) REFERENCES products (product_id)
+        )
+    """,
+    
+    "prescribers": """
+        CREATE TABLE IF NOT EXISTS prescribers (
+            prescriber_id TEXT PRIMARY KEY,
+            name TEXT NOT NULL,
+            specialty TEXT NOT NULL,
+            practice_type TEXT NOT NULL,
+            territory_id TEXT NOT NULL,
+            decile INTEGER NOT NULL,
+            FOREIGN KEY (territory_id) REFERENCES territories (territory_id)
+        )
+    """,
+    
+    "pharmacies": """
+        CREATE TABLE IF NOT EXISTS pharmacies (
+            pharmacy_id TEXT PRIMARY KEY,
+            name TEXT NOT NULL,
+            address TEXT NOT NULL,
+            territory_id TEXT NOT NULL,
+            pharmacy_type TEXT NOT NULL,
+            monthly_rx_volume INTEGER NOT NULL,
+            FOREIGN KEY (territory_id) REFERENCES territories (territory_id)
+        )
+    """,
+    
+    "distribution_centers": """
+        CREATE TABLE IF NOT EXISTS distribution_centers (
+            dc_id TEXT PRIMARY KEY,
+            dc_name TEXT NOT NULL,
+            region_id TEXT NOT NULL,
+            inventory_capacity INTEGER NOT NULL,
+            FOREIGN KEY (region_id) REFERENCES regions (region_id)
+        )
+    """,
+    
+    "sales": """
+        CREATE TABLE IF NOT EXISTS sales (
+            sale_id INTEGER PRIMARY KEY,
+            sale_date DATE NOT NULL,
+            product_id TEXT NOT NULL,
+            region_id TEXT NOT NULL,
+            territory_id TEXT NOT NULL,
+            prescriber_id TEXT NOT NULL,
+            pharmacy_id TEXT NOT NULL,
+            units_sold INTEGER NOT NULL,
+            revenue REAL NOT NULL,
+            cost REAL NOT NULL,
+            margin REAL NOT NULL,
+            FOREIGN KEY (product_id) REFERENCES products (product_id),
+            FOREIGN KEY (region_id) REFERENCES regions (region_id),
+            FOREIGN KEY (territory_id) REFERENCES territories (territory_id),
+            FOREIGN KEY (prescriber_id) REFERENCES prescribers (prescriber_id),
+            FOREIGN KEY (pharmacy_id) REFERENCES pharmacies (pharmacy_id)
+        )
+    """,
+    
+    "marketing_campaigns": """
+        CREATE TABLE IF NOT EXISTS marketing_campaigns (
+            campaign_id INTEGER PRIMARY KEY,
+            campaign_name TEXT NOT NULL,
+            start_date DATE NOT NULL,
+            end_date DATE NOT NULL,
+            product_id TEXT NOT NULL,
+            campaign_type TEXT NOT NULL,
+            target_audience TEXT NOT NULL,
+            channels TEXT NOT NULL,
+            budget REAL NOT NULL,
+            spend REAL NOT NULL,
+            FOREIGN KEY (product_id) REFERENCES products (product_id)
+        )
+    """,
+    
+    "market_events": """
+        CREATE TABLE IF NOT EXISTS market_events (
+            event_id INTEGER PRIMARY KEY,
+            event_date DATE NOT NULL,
+            event_type TEXT NOT NULL,
+            description TEXT NOT NULL,
+            affected_products TEXT NOT NULL,
+            affected_regions TEXT NOT NULL,
+            impact_score REAL NOT NULL
+        )
+    """,
+    
+    "sales_targets": """
+        CREATE TABLE IF NOT EXISTS sales_targets (
+            target_id INTEGER PRIMARY KEY,
+            product_id TEXT NOT NULL,
+            region_id TEXT NOT NULL,
+            period TEXT NOT NULL,
+            target_units REAL NOT NULL,
+            target_revenue REAL NOT NULL,
+            FOREIGN KEY (product_id) REFERENCES products (product_id),
+            FOREIGN KEY (region_id) REFERENCES regions (region_id)
+        )
+    """,
+    
+    "inventory": """
+        CREATE TABLE IF NOT EXISTS inventory (
+            inventory_id INTEGER PRIMARY KEY,
+            product_id TEXT NOT NULL,
+            dc_id TEXT NOT NULL,
+            date DATE NOT NULL,
+            units_available INTEGER NOT NULL,
+            units_allocated INTEGER NOT NULL,
+            units_in_transit INTEGER NOT NULL,
+            days_of_supply REAL NOT NULL,
+            FOREIGN KEY (product_id) REFERENCES products (product_id),
+            FOREIGN KEY (dc_id) REFERENCES distribution_centers (dc_id)
+        )
+    """,
+    
+    "external_factors": """
+        CREATE TABLE IF NOT EXISTS external_factors (
+            factor_id INTEGER PRIMARY KEY,
+            date DATE NOT NULL,
+            region_id TEXT NOT NULL,
+            factor_type TEXT NOT NULL,
+            factor_value REAL NOT NULL,
+            description TEXT NOT NULL,
+            FOREIGN KEY (region_id) REFERENCES regions (region_id)
+        )
+    """
+}
+
+# Indexes for optimizing queries
+INDEX_DEFINITIONS = [
+    "CREATE INDEX IF NOT EXISTS idx_sales_date ON sales (sale_date)",
+    "CREATE INDEX IF NOT EXISTS idx_sales_product ON sales (product_id)",
+    "CREATE INDEX IF NOT EXISTS idx_sales_region ON sales (region_id)",
+    "CREATE INDEX IF NOT EXISTS idx_sales_territory ON sales (territory_id)",
+    "CREATE INDEX IF NOT EXISTS idx_inventory_date ON inventory (date)",
+    "CREATE INDEX IF NOT EXISTS idx_inventory_product ON inventory (product_id)",
+    "CREATE INDEX IF NOT EXISTS idx_marketing_dates ON marketing_campaigns (start_date, end_date)",
+    "CREATE INDEX IF NOT EXISTS idx_external_factors_date ON external_factors (date)",
+    "CREATE INDEX IF NOT EXISTS idx_external_factors_region ON external_factors (region_id)"
+]
+
+def create_schema(db_path: str = "data/pharma_db.sqlite") -> None:
+    """Create database schema in the SQLite database"""
+    conn = sqlite3.connect(db_path)
+    cursor = conn.cursor()
+    
+    # Create tables
+    for table_name, schema_sql in SCHEMA_DEFINITIONS.items():
+        cursor.execute(schema_sql)
+    
+    # Create indexes
+    for index_sql in INDEX_DEFINITIONS:
+        cursor.execute(index_sql)
+    
+    conn.commit()
+    conn.close()
+    
+    print(f"Database schema created in {db_path}")
+
+def get_table_info(db_path: str = "data/pharma_db.sqlite") -> Dict[str, List[Dict[str, Any]]]:
+    """Get information about all tables in the database"""
+    conn = sqlite3.connect(db_path)
+    cursor = conn.cursor()
+    
+    # Get list of tables
+    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
+    tables = cursor.fetchall()
+    
+    table_info = {}
+    
+    for table in tables:
+        table_name = table[0]
+        
+        # Get column information
+        cursor.execute(f"PRAGMA table_info({table_name})")
+        columns = cursor.fetchall()
+        
+        # Format column information
+        column_info = []
+        for col in columns:
+            column_info.append({
+                "name": col[1],
+                "type": col[2],
+                "notnull": col[3],
+                "default": col[4],
+                "primary_key": col[5]
+            })
+        
+        table_info[table_name] = column_info
+    
+    conn.close()
+    return table_info
+
+if __name__ == "__main__":
+    # Create the database schema when run directly
+    create_schema()
+    
+    # Print table information
+    table_info = get_table_info()
+    for table_name, columns in table_info.items():
+        print(f"Table: {table_name}")
+        for col in columns:
+            print(f"  - {col['name']} ({col['type']})")
+        print()