Shared Data Schemas

{
  "league_name": "string",
  "sport_category": "string",
  "contact_email": "string",
  "contact_phone": "string",
  "website_url": "string",
  "founded_year": "integer",
  "headquarters_location": "string",
  "number_of_teams": "integer",
  "season_structure": "string",
  "total_annual_events": "integer",
  "average_attendance": "integer",
  "current_sponsorships": "array",
  "betting_market_interest": "boolean",
  "data_availability": "string",
  "additional_requirements": "string"
}

from data_layer.shared import LeagueQuestionnaireSchema
 
# Create validated instance
questionnaire = LeagueQuestionnaireSchema(
    league_name="Canadian Premier League",
    sport_category="Soccer",
    contact_email="partnerships@cpl.ca",
    founded_year=2017,
    number_of_teams=8,
    season_structure="Spring-Fall",
    total_annual_events=117,
    betting_market_interest=True
)
 
# Access with type safety
print(f"Processing application for {questionnaire.league_name}")
print(f"Events per year: {questionnaire.total_annual_events}")

import { LeagueQuestionnaire } from '@/data-layer/types/league_questionnaire';
 
// Type-safe interface
const questionnaire: LeagueQuestionnaire = {
  league_name: "Canadian Premier League",
  sport_category: "Soccer",
  contact_email: "partnerships@cpl.ca",
  founded_year: 2017,
  number_of_teams: 8,
  betting_market_interest: true
};

{
  "contract_id": "string",
  "league_name": "string",
  "tier": "string",
  "base_price": "number",
  "revenue_share_percentage": "number",
  "contract_duration_months": "integer",
  "payment_schedule": "string",
  "data_delivery_requirements": "object",
  "marketing_obligations": "array",
  "termination_clauses": "object",
  "renewal_options": "object"
}

from data_layer.shared import ContractTermsSchema
 
contract = ContractTermsSchema(
    contract_id="CPL-2025-001",
    league_name="Canadian Premier League",
    tier="premium",
    base_price=250000,
    revenue_share_percentage=15.0,
    contract_duration_months=12,
    payment_schedule="quarterly"
)

{
  "league_name": "string",
  "overall_score": "number",
  "demographics_score": "number",
  "seasonality_score": "number",
  "market_maturity_score": "number",
  "data_quality_score": "number",
  "recommended_tier": "string",
  "estimated_annual_value": "number",
  "confidence_level": "number",
  "risk_factors": "array",
  "growth_potential": "string"
}

from data_layer.shared import TierClassificationSchema
 
classification = TierClassificationSchema(
    league_name="FWT",
    overall_score=7.8,
    demographics_score=8.5,
    seasonality_score=6.2,
    recommended_tier="tier_2",
    estimated_annual_value=125000,
    confidence_level=0.85,
    risk_factors=["seasonal_sport", "limited_data"],
    growth_potential="high"
)

{
  "package_id": "string",
  "league_name": "string",
  "proposed_terms": "object",
  "counter_offers": "array",
  "negotiation_history": "array",
  "current_status": "string",
  "deadline": "string",
  "decision_makers": "array",
  "special_conditions": "array"
}

from pydantic import ValidationError
from data_layer.shared import LeagueQuestionnaireSchema
 
try:
    questionnaire = LeagueQuestionnaireSchema(
        league_name="",  # Invalid: empty string
        founded_year=1800,  # Invalid: too old
        contact_email="invalid-email"  # Invalid: bad format
    )
except ValidationError as e:
    print(f"Validation errors: {e}")

# Custom validators ensure business logic
class LeagueQuestionnaireSchema(BaseModel):
    league_name: str = Field(..., min_length=2, max_length=100)
    founded_year: int = Field(..., ge=1800, le=2025)
    contact_email: EmailStr
    number_of_teams: int = Field(..., gt=0, le=1000)
 
    @field_validator('season_structure')
    @classmethod
    def validate_season_structure(cls, v):
        valid_structures = ['Spring-Fall', 'Fall-Spring', 'Year-round', 'Seasonal']
        if v not in valid_structures:
            raise ValueError(f'Season structure must be one of: {valid_structures}')
        return v

# Auto-generated Pydantic models
from data_layer.shared.python.league_questionnaire import LeagueQuestionnaire
 
questionnaire = LeagueQuestionnaire(
    league_name="Test League",
    sport_category="Soccer"
)

// Auto-generated TypeScript interfaces
import { LeagueQuestionnaire } from '@/data-layer/types';
 
const questionnaire: LeagueQuestionnaire = {
  league_name: "Test League",
  sport_category: "Soccer"
};

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "league_name": {
      "type": "string",
      "minLength": 2,
      "maxLength": 100
    },
    "sport_category": {
      "type": "string",
      "enum": ["Soccer", "Basketball", "Baseball", "Hockey", "Tennis"]
    }
  },
  "required": ["league_name", "sport_category"]
}

from fastapi import APIRouter, HTTPException
from data_layer.shared import LeagueQuestionnaireSchema
 
router = APIRouter()
 
@router.post("/questionnaire/submit")
async def submit_questionnaire(data: dict):
    try:
        questionnaire = LeagueQuestionnaireSchema(**data)
 
        # Process with validated data
        result = await process_league_application(questionnaire)
 
        return {"status": "submitted", "id": result.id}
 
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())

import { useState } from 'react';
import { LeagueQuestionnaire } from '@/data-layer/types';
 
export function LeagueQuestionnaireForm() {
  const [formData, setFormData] = useState<Partial<LeagueQuestionnaire>>({});
 
  const handleSubmit = async () => {
    // Type-safe form submission
    const response = await fetch('/api/questionnaire/submit', {
      method: 'POST',
      body: JSON.stringify(formData)
    });
 
    if (response.ok) {
      // Success handling
    }
  };
 
  return (
    <form onSubmit={handleSubmit}>
      {/* Form fields with full TypeScript support */}
    </form>
  );
}

from crewai.flow.flow import Flow, start
from data_layer.shared import LeagueQuestionnaireSchema, TierClassificationSchema
 
class PartnershipFlow(Flow):
    @start()
    def analyze_league(self):
        # Validate input
        questionnaire = LeagueQuestionnaireSchema(**self.state.input_data)
 
        # AI-powered tier classification
        classification = await self.analyze_tier(questionnaire)
 
        # Store validated results
        self.state.questionnaire = questionnaire
        self.state.classification = classification
 
        return classification

# Support multiple schema versions
class LeagueQuestionnaireSchema(BaseModel):
    league_name: str
    sport_category: str
 
    # New fields with defaults for backward compatibility
    contact_email: Optional[str] = None  # Added in v1.1
    betting_market_interest: bool = False  # Added in v1.2

import pytest
from data_layer.shared import LeagueQuestionnaireSchema
 
def test_valid_questionnaire():
    data = {
        "league_name": "Test League",
        "sport_category": "Soccer",
        "contact_email": "test@example.com"
    }
 
    questionnaire = LeagueQuestionnaireSchema(**data)
    assert questionnaire.league_name == "Test League"
 
def test_invalid_questionnaire():
    data = {
        "league_name": "",  # Invalid: empty string
        "sport_category": "Invalid Sport"
    }
 
    with pytest.raises(ValidationError):
        LeagueQuestionnaireSchema(**data)

def test_full_data_flow():
    # Test complete flow from input to processing
    raw_data = load_test_questionnaire()
    questionnaire = LeagueQuestionnaireSchema(**raw_data)
 
    # Process through business logic
    result = process_questionnaire(questionnaire)
 
    # Verify output schema
    assert isinstance(result.classification, TierClassificationSchema)
    assert isinstance(result.contract, ContractTermsSchema)

from data_layer.shared.registry import SchemaRegistry
 
registry = SchemaRegistry()
schemas = registry.list_schemas()  # Returns all available schemas
questionnaire_schema = registry.get_schema('league_questionnaire')