Database & Models

class Model(Base, sqlmodel.SQLModel):
    """
    Base class to define a table in the database.
    
    Inherits from both Reflex Base and SQLModel, providing automatic
    table creation, relationship support, and seamless integration with
    the Reflex framework and FastAPI/Pydantic ecosystem.
    
    The default primary key field is automatically dropped if any other
    field is marked as primary key.
    """
    
    # The primary key for the table (auto-dropped if custom primary key exists)
    id: int | None = sqlmodel.Field(default=None, primary_key=True)
    
    def __init_subclass__(cls):
        """
        Drop the default primary key field if any primary key field is defined.
        
        Automatically handles custom primary key configurations by removing
        the default 'id' field when other fields are marked as primary keys.
        """
        ...
    
    @classmethod
    def _dict_recursive(cls, value: Any):
        """
        Recursively serialize the relationship object(s).
        
        Args:
            value: The value to serialize
        
        Returns:
            The serialized value
        """
        ...
    
    def dict(self, **kwargs):
        """
        Convert the object to a dictionary.
        
        Includes both base fields and relationship data when available.
        Handles SQLModel relationships that don't appear in __fields__.
        
        Args:
            kwargs: Ignored but needed for compatibility
        
        Returns:
            The object as a dictionary with fields and relationships
        """
        ...

class ModelRegistry:
    """
    Registry for all models in the application.
    
    Manages model registration and metadata collection for database
    operations, migrations, and table creation.
    """
    
    models: ClassVar[set[SQLModelOrSqlAlchemy]] = set()
    
    @classmethod
    def register(cls, model: SQLModelOrSqlAlchemy):
        """
        Register a model. Can be used directly or as a decorator.
        
        Args:
            model: The model to register
        
        Returns:
            The model passed in as an argument (allows decorator usage)
        """
        ...
    
    @classmethod
    def get_models(cls, include_empty: bool = False) -> set[SQLModelOrSqlAlchemy]:
        """
        Get registered models.
        
        Args:
            include_empty: If True, include models with empty metadata
        
        Returns:
            The registered models
        """
        ...
    
    @classmethod
    def get_metadata(cls) -> sqlalchemy.MetaData:
        """
        Get the metadata for all registered models.
        
        Returns:
            Combined metadata from all registered models
        """
        ...
    
    @classmethod
    def filter(cls, **conditions) -> list[Model]:
        """
        Filter model instances by field conditions.
        
        Args:
            **conditions: Field name to value mappings for filtering
        
        Returns:
            List of model instances matching conditions
        """
        ...
    
    @classmethod
    def count(cls) -> int:
        """
        Count total number of instances in database.
        
        Returns:
            Total count of records for this model
        """
        ...

import reflex as rx
from datetime import datetime

class User(rx.Model, table=True):
    username: str
    email: str
    created_at: datetime = datetime.now()
    is_active: bool = True

class Post(rx.Model, table=True):
    title: str
    content: str
    author_id: int  # Foreign key to User
    published: bool = False
    
    # Relationship
    author: User = rx.Relationship(back_populates="posts")

# Usage
user = User.create(username="john", email="john@example.com")
post = Post.create(title="Hello World", content="My first post", author_id=user.id)

def session() -> Session:
    """
    Get database session for synchronous operations.
    
    Provides SQLAlchemy session with automatic connection management,
    transaction handling, and cleanup for database operations.
    
    Returns:
        SQLAlchemy Session instance for database operations
    """
    ...

def asession() -> AsyncSession:
    """
    Get async database session for high-performance operations.
    
    Provides async SQLAlchemy session for non-blocking database
    operations, ideal for high-concurrency applications.
    
    Returns:
        SQLAlchemy AsyncSession for async database operations
    """
    ...

# Synchronous database operations
with rx.session() as db:
    users = db.query(User).filter(User.is_active == True).all()
    new_user = User(username="alice", email="alice@example.com")
    db.add(new_user)
    db.commit()

# Asynchronous database operations  
async def get_users_async():
    async with rx.asession() as db:
        result = await db.execute(select(User).where(User.is_active == True))
        return result.scalars().all()

class DBConfig:
    """
    Database configuration with PostgreSQL optimization.
    
    Manages database connections, migration settings, and performance
    tuning for production and development environments.
    """
    
    db_url: str
    """Database connection URL (PostgreSQL, SQLite, MySQL, etc.)."""
    
    engine: Engine | None = None
    """SQLAlchemy engine instance (auto-created if None)."""
    
    migrate: bool = True
    """Whether to run automatic database migrations on startup."""
    
    pool_size: int = 10
    """Number of connections to maintain in connection pool."""
    
    max_overflow: int = 20
    """Maximum overflow connections beyond pool_size."""
    
    pool_timeout: int = 30
    """Timeout in seconds for getting connection from pool."""
    
    pool_recycle: int = 3600
    """Seconds before connection is recreated (prevents stale connections)."""
    
    echo: bool = False
    """Whether to log all SQL statements (useful for debugging)."""
    
    echo_pool: bool = False
    """Whether to log connection pool operations."""
    
    def __init__(self, db_url: str, **kwargs) -> None:
        """
        Initialize database configuration.
        
        Args:
            db_url: Database connection URL
            **kwargs: Additional configuration options to override defaults
        """
        ...
    
    def create_engine(self) -> Engine:
        """
        Create SQLAlchemy engine with optimized settings.
        
        Returns:
            Configured SQLAlchemy Engine instance
        """
        ...
    
    def create_tables(self) -> None:
        """
        Create all database tables based on model definitions.
        
        Uses SQLAlchemy metadata to create tables for all registered
        models, handling dependencies and foreign key constraints.
        """
        ...
    
    def run_migrations(self) -> None:
        """
        Run database migrations to update schema.
        
        Applies pending Alembic migrations to bring database schema
        up to date with current model definitions.
        """
        ...

import reflex as rx

config = rx.Config(
    app_name="myapp",
    db_config=rx.DBConfig(
        db_url="postgresql://user:pass@localhost/mydb",
        pool_size=20,
        max_overflow=30,
        echo=False  # Set to True for SQL debugging
    )
)

# Model querying patterns
class QueryMixin:
    """Query building utilities for database models."""
    
    @classmethod
    def where(cls, *conditions) -> Query:
        """
        Build query with WHERE conditions.
        
        Args:
            *conditions: SQLAlchemy filter conditions
        
        Returns:
            Query object with applied conditions
        """
        ...
    
    @classmethod
    def join(cls, *models) -> Query:
        """
        Build query with JOIN operations.
        
        Args:
            *models: Model classes to join with
        
        Returns:
            Query object with joins applied
        """
        ...
    
    @classmethod
    def order_by(cls, *fields) -> Query:
        """
        Build query with ORDER BY clause.
        
        Args:
            *fields: Fields to order by (use .desc() for descending)
        
        Returns:
            Query object with ordering applied
        """
        ...
    
    @classmethod
    def limit(cls, count: int) -> Query:
        """
        Limit query results to specific count.
        
        Args:
            count: Maximum number of results to return
        
        Returns:
            Query object with limit applied
        """
        ...
    
    @classmethod
    def offset(cls, count: int) -> Query:
        """
        Skip specified number of results.
        
        Args:
            count: Number of results to skip
        
        Returns:
            Query object with offset applied
        """
        ...

from sqlalchemy.orm import relationship

# Relationship patterns for models
class User(rx.Model, table=True):
    id: int = Field(primary_key=True)
    username: str
    email: str
    
    # One-to-many relationship
    posts: list["Post"] = relationship("Post", back_populates="author")
    profile: "UserProfile" = relationship("UserProfile", back_populates="user", uselist=False)

class Post(rx.Model, table=True):
    id: int = Field(primary_key=True)
    title: str
    content: str
    author_id: int = Field(foreign_key="user.id")
    
    # Many-to-one relationship
    author: User = relationship("User", back_populates="posts")
    
    # Many-to-many relationship
    tags: list["Tag"] = relationship("Tag", secondary="post_tags", back_populates="posts")

class Tag(rx.Model, table=True):
    id: int = Field(primary_key=True)
    name: str
    
    # Many-to-many relationship
    posts: list[Post] = relationship("Post", secondary="post_tags", back_populates="tags")

# Association table for many-to-many
class PostTag(rx.Model, table=True):
    post_id: int = Field(foreign_key="post.id", primary_key=True)
    tag_id: int = Field(foreign_key="tag.id", primary_key=True)

def create_migration(message: str) -> None:
    """
    Create new database migration file.
    
    Generates Alembic migration file comparing current model
    definitions against existing database schema.
    
    Args:
        message: Descriptive message for the migration
    """
    ...

def run_migrations() -> None:
    """
    Apply pending migrations to database.
    
    Runs all pending Alembic migrations to bring database
    schema up to date with current model definitions.
    """
    ...

def rollback_migration(revision: str = None) -> None:
    """
    Rollback database to previous migration.
    
    Args:
        revision: Specific revision to rollback to (latest if None)
    """
    ...

def get_migration_status() -> dict:
    """
    Get current migration status and pending changes.
    
    Returns:
        Dictionary with current revision and pending migration info
    """
    ...

import reflex as rx
from datetime import datetime
from typing import Optional

class User(rx.Model, table=True):
    # Primary key (automatic)
    id: int = Field(primary_key=True)
    
    # Required fields
    username: str = Field(unique=True, index=True)
    email: str = Field(unique=True)
    
    # Optional fields
    full_name: Optional[str] = None
    is_active: bool = True
    
    # Timestamp fields (automatic)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

class UserState(rx.State):
    users: list[User] = []
    
    def load_active_users(self):
        with rx.session() as db:
            self.users = db.query(User).filter(
                User.is_active == True
            ).order_by(User.created_at.desc()).all()
    
    def search_users(self, query: str):
        with rx.session() as db:
            self.users = db.query(User).filter(
                User.username.contains(query) |
                User.full_name.contains(query)
            ).limit(10).all()

class AsyncUserState(rx.State):
    users: list[User] = []
    
    async def load_users_async(self):
        async with rx.asession() as db:
            result = await db.execute(
                select(User).where(User.is_active == True)
                .order_by(User.created_at.desc())
            )
            self.users = result.scalars().all()

from typing import Any, Dict, List, Optional, Union
from sqlalchemy import Engine, Column
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime

# Database Types
ModelInstance = Any  # Database model instance
QueryResult = List[ModelInstance]  # Query result list
ModelDict = Dict[str, Any]  # Model as dictionary

# Session Types
DBSession = Session  # Synchronous database session
AsyncDBSession = AsyncSession  # Asynchronous database session

# Configuration Types
DatabaseURL = str  # Database connection URL
EngineConfig = Dict[str, Any]  # SQLAlchemy engine configuration

# Field Types
PrimaryKey = int  # Primary key field type
ForeignKey = int  # Foreign key field type
Timestamp = datetime  # Timestamp field type

# Query Types
WhereCondition = Any  # SQLAlchemy WHERE condition
JoinCondition = Any  # SQLAlchemy JOIN condition
OrderByField = Any  # SQLAlchemy ORDER BY field

# Migration Types
RevisionID = str  # Alembic revision identifier
MigrationMessage = str  # Migration description