YeboHub

Appearance

Okia Database Models ​

SQLAlchemy models in /app/db/models.py.

Enums ​

python
from enum import Enum

class InterviewStatus(str, Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    PAUSED = "paused"
    COMPLETED = "completed"
    EXPIRED = "expired"

class MessageRole(str, Enum):
    OKIA = "okia"
    CANDIDATE = "candidate"

class MessageType(str, Enum):
    GREETING = "greeting"
    QUESTION = "question"
    ANSWER = "answer"
    CLARIFICATION = "clarification"
    CLOSING = "closing"
    SYSTEM = "system"

OkiaSession ​

Main interview session model.

python
class OkiaSession(Base):
    __tablename__ = "okia_sessions"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    session_token = Column(String(64), unique=True, nullable=False, index=True)
    
    # Candidate info
    candidate_name = Column(String(255), nullable=False)
    candidate_phone = Column(String(20), nullable=False)
    candidate_email = Column(String(255))
    
    # YeboJobs references
    user_id = Column(String(50))  # YeboJobs user CUID
    application_id = Column(String(50))  # YeboJobs application CUID
    
    # CV data
    cv_url = Column(Text)
    cv_text = Column(Text)  # Extracted text for AI context
    
    # Job context
    job_category = Column(String(100), nullable=False)
    job_title = Column(String(255))
    job_description = Column(Text)
    experience_level = Column(String(20))  # entry, mid, senior
    
    # Interview settings
    max_questions = Column(Integer, default=10)
    time_limit_minutes = Column(Integer, default=15)
    question_count = Column(Integer, default=0)
    
    # Status
    status = Column(SQLEnum(InterviewStatus), default=InterviewStatus.PENDING)
    
    # Timestamps
    started_at = Column(DateTime(timezone=True))
    completed_at = Column(DateTime(timezone=True))
    paused_at = Column(DateTime(timezone=True))
    total_paused_seconds = Column(Integer, default=0)
    expires_at = Column(DateTime(timezone=True), nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    # Interview context (JSON for dynamic data)
    interview_context = Column(JSONB, default={
        "questions_asked": 0,
        "questions_remaining": 10,
        "topics_covered": [],
        "current_question": None,
    })
    
    # Webhook configuration
    webhook_url = Column(Text)
    webhook_metadata = Column(JSONB)
    webhook_secret = Column(String(255))
    
    # SMS tracking
    sms_sent = Column(Boolean, default=False)
    sms_sent_at = Column(DateTime(timezone=True))
    
    # Relationships
    messages = relationship("OkiaMessage", back_populates="session", cascade="all, delete-orphan")
    scores = relationship("OkiaScore", back_populates="session", cascade="all, delete-orphan")
    report = relationship("OkiaReport", back_populates="session", uselist=False, cascade="all, delete-orphan")

Indexes ​

python
Index("ix_okia_sessions_user_id", OkiaSession.user_id)
Index("ix_okia_sessions_status", OkiaSession.status)
Index("ix_okia_sessions_created_at", OkiaSession.created_at)

OkiaMessage ​

Individual messages in an interview.

python
class OkiaMessage(Base):
    __tablename__ = "okia_messages"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    session_id = Column(UUID(as_uuid=True), ForeignKey("okia_sessions.id", ondelete="CASCADE"), nullable=False)
    
    # Message content
    role = Column(SQLEnum(MessageRole), nullable=False)
    message_type = Column(SQLEnum(MessageType), nullable=False)
    content = Column(Text, nullable=False)
    
    # Question metadata (for QUESTION type)
    question_number = Column(Integer)
    question_category = Column(String(50))  # technical, behavioral, situational
    expected_elements = Column(Text)  # JSON array of expected answer elements
    
    # Scoring status
    is_scored = Column(Boolean, default=False)
    
    # Ordering
    sequence_number = Column(Integer, nullable=False)
    
    # Timestamps
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    # Relationships
    session = relationship("OkiaSession", back_populates="messages")
    score = relationship("OkiaScore", back_populates="message", uselist=False)

Indexes ​

python
Index("ix_okia_messages_session_id", OkiaMessage.session_id)
Index("ix_okia_messages_sequence", OkiaMessage.session_id, OkiaMessage.sequence_number)

OkiaScore ​

Score for each candidate answer.

python
class OkiaScore(Base):
    __tablename__ = "okia_scores"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    session_id = Column(UUID(as_uuid=True), ForeignKey("okia_sessions.id", ondelete="CASCADE"), nullable=False)
    message_id = Column(UUID(as_uuid=True), ForeignKey("okia_messages.id", ondelete="CASCADE"), nullable=False)
    
    # Individual scores (0-100)
    relevance_score = Column(Integer, nullable=False)
    depth_score = Column(Integer, nullable=False)
    clarity_score = Column(Integer, nullable=False)
    technical_score = Column(Integer)  # Optional, for technical questions
    
    # Weighted composite
    composite_score = Column(Integer, nullable=False)
    
    # AI-generated feedback
    scoring_rationale = Column(Text)  # Why this score
    strengths = Column(JSONB, default=[])  # What was good
    improvements = Column(JSONB, default=[])  # What could be better
    notable_quote = Column(Text)  # Memorable part of answer
    
    # Timestamps
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    # Relationships
    session = relationship("OkiaSession", back_populates="scores")
    message = relationship("OkiaMessage", back_populates="score")

OkiaReport ​

Final interview report.

python
class OkiaReport(Base):
    __tablename__ = "okia_reports"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    session_id = Column(UUID(as_uuid=True), ForeignKey("okia_sessions.id", ondelete="CASCADE"), unique=True, nullable=False)
    
    # Category scores (0-100)
    technical_score = Column(Integer, nullable=False)
    communication_score = Column(Integer, nullable=False)
    problem_solving_score = Column(Integer, nullable=False)
    cultural_fit_score = Column(Integer, nullable=False)
    
    # Overall assessment
    overall_score = Column(Integer, nullable=False)  # 0-100
    grade = Column(String(2), nullable=False)  # A+, A, B+, B, C, D, F
    recommendation = Column(String(20), nullable=False)  # strong_hire, hire, consider, no_hire
    
    # Detailed analysis
    executive_summary = Column(Text)
    strengths_analysis = Column(JSONB, default=[])
    weaknesses_analysis = Column(JSONB, default=[])
    
    # Technical assessment
    technical_assessment = Column(Text)
    behavioral_assessment = Column(Text)
    
    # Skills
    demonstrated_skills = Column(JSONB, default=[])
    skill_gaps = Column(JSONB, default=[])
    
    # Highlights & concerns
    interview_highlights = Column(JSONB, default=[])  # [{question, highlight}]
    red_flags = Column(JSONB, default=[])
    
    # Recommendations
    follow_up_questions = Column(JSONB, default=[])
    
    # PDF storage
    pdf_url = Column(Text)
    
    # Timestamps
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    # Relationships
    session = relationship("OkiaSession", back_populates="report")

Database Session ​

Async database connection setup:

python
# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=5,
    max_overflow=10,
)

async_session_maker = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_db() -> AsyncSession:
    """Dependency for FastAPI routes."""
    async with async_session_maker() as session:
        try:
            yield session
        finally:
            await session.close()

Migrations ​

Using Alembic for migrations:

bash
# Create migration
alembic revision --autogenerate -m "Add new field"

# Apply migrations
alembic upgrade head

# Rollback
alembic downgrade -1

Example Migration ​

python
# alembic/versions/xxx_add_sms_tracking.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('okia_sessions', sa.Column('sms_sent', sa.Boolean(), default=False))
    op.add_column('okia_sessions', sa.Column('sms_sent_at', sa.DateTime(timezone=True)))

def downgrade():
    op.drop_column('okia_sessions', 'sms_sent_at')
    op.drop_column('okia_sessions', 'sms_sent')

Query Examples ​

Get session with all relations ​

python
from sqlalchemy import select
from sqlalchemy.orm import selectinload

async def get_session_by_token(db: AsyncSession, token: str) -> Optional[OkiaSession]:
    result = await db.execute(
        select(OkiaSession)
        .where(OkiaSession.session_token == token)
        .options(
            selectinload(OkiaSession.messages),
            selectinload(OkiaSession.scores),
            selectinload(OkiaSession.report),
        )
    )
    return result.scalar_one_or_none()

Get user's completed sessions ​

python
async def get_user_completed_sessions(
    db: AsyncSession, 
    user_id: str,
    limit: int = 10,
) -> List[OkiaSession]:
    result = await db.execute(
        select(OkiaSession)
        .where(
            OkiaSession.user_id == user_id,
            OkiaSession.status == InterviewStatus.COMPLETED,
        )
        .order_by(OkiaSession.completed_at.desc())
        .limit(limit)
    )
    return result.scalars().all()

Calculate average score ​

python
from sqlalchemy import func

async def get_user_average_score(db: AsyncSession, user_id: str) -> Optional[float]:
    result = await db.execute(
        select(func.avg(OkiaReport.overall_score))
        .join(OkiaSession)
        .where(OkiaSession.user_id == user_id)
    )
    return result.scalar()

One chat. Everything done.

© 2026 Omevision Ltd.