Models to GraphQL

🚀 Automatic GraphQL schema generation from SQLAlchemy models with relationship support, typed filtering, and Django REST Framework-style ViewSets

✨ Features

🔄 Automatic generation of GraphQL types, queries and mutations from SQLAlchemy models
🔗 Smart relationships - automatic resolution of relationships between models
🎯 Typed filters - Django-style filtering with operation support (icontains, gte, in, etc.)
📄 Pagination - built-in limit/offset and page-based pagination support
🎨 ViewSets - Django REST Framework approach for API organization
🚀 Apollo Federation - ready-to-use microservice architecture support
📝 TypeScript export - auto-generation of TypeScript types and GraphQL operations
⚡ Async/Await - full SQLAlchemy 2.0 async support
🛠 Customization - flexible hooks and overrides for any needs

🔧 Installation

# Basic installation
pip install models-to-graphql

# With Apollo Federation support
pip install models-to-graphql[federation]

# For development
pip install models-to-graphql[dev]

# With database drivers
pip install models-to-graphql[postgres]  # PostgreSQL
pip install models-to-graphql[mysql]     # MySQL
pip install models-to-graphql[sqlite]    # SQLite

🚀 Quick Start

1. Define SQLAlchemy models

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from enum import Enum

Base = declarative_base()

class StatusEnum(Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True)
    status = Column(Enum(StatusEnum), default=StatusEnum.ACTIVE)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships
    author = relationship("User", back_populates="posts")

2. Create ViewSets

from models_to_graphql import AutoGraphQLViewSet

class UserViewSet(AutoGraphQLViewSet):
    model = User
    exclude_fields = ['password_hash']  # Exclude sensitive data
    
    # Custom creation logic
    async def prepare_create_data(self, data, input_data):
        data = await super().prepare_create_data(data, input_data)
        # Add additional logic
        data['created_by'] = 'system'
        return data

class PostViewSet(AutoGraphQLViewSet):
    model = Post
    
    # Custom hooks
    async def post_create(self, obj, input_data):
        # Notify about new post
        await notify_new_post(obj)
        return await super().post_create(obj, input_data)

3. Create GraphQL schema

from models_to_graphql import create_graphql_schema_from_viewsets

# Schema creation
schema = create_graphql_schema_from_viewsets([
    ("user", UserViewSet),
    ("post", PostViewSet),
], with_relationships=True)

# Usage with FastAPI
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

app = FastAPI()
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

4. Done! 🎉

Now you have a full-featured GraphQL API with:

# Queries
query {
  users(
    filters: { 
      name: { icontains: "john" }
      status: { exact: "active" }
    }
    limit: 10
    offset: 0
  ) {
    items {
      id
      name
      email
      posts {
        id
        title
        content
      }
    }
    pagination {
      totalCount
      hasNext
      hasPrevious
    }
  }
}

# Mutations
mutation {
  create_user(input: {
    name: "John Doe"
    email: "john@example.com"
  }) {
    id
    name
    email
  }
}

📚 Detailed Documentation

ViewSets - Architecture Foundation

AutoGraphQLViewSet provides a Django REST Framework approach:

class ProductViewSet(AutoGraphQLViewSet):
    model = Product
    
    # Field settings
    exclude_fields = ['internal_data', 'secret_key']
    filter_exclude_fields = ['password']
    
    # Method settings
    default_methods = ["list", "retrieve", "create", "update", "delete"]
    
    # Name settings
    mutation_prefix = "product"  # create_product, update_product
    query_prefix = "product"     # products, product
    
    # Custom filtering
    async def filter_queryset(self, queryset, filter_params, session):
        queryset = await super().filter_queryset(queryset, filter_params, session)
        # Add additional filters
        return queryset.where(Product.is_active == True)
    
    # Lifecycle hooks
    async def pre_create(self, session, input_data):
        # Validation before creation
        if await self.email_exists(input_data.email):
            raise ValueError("Email already exists")
    
    async def post_create(self, obj, input_data):
        # Actions after creation
        await self.send_welcome_email(obj.email)
        return obj

Typed Filtering

Automatically generated typed filters for each field type:

# For string fields
class StringFilterInput:
    exact: Optional[str] = None
    icontains: Optional[str] = None
    contains: Optional[str] = None
    startswith: Optional[str] = None
    endswith: Optional[str] = None
    in_: Optional[List[str]] = None
    isnull: Optional[bool] = None

# For numeric fields
class IntFilterInput:
    exact: Optional[int] = None
    gt: Optional[int] = None
    gte: Optional[int] = None
    lt: Optional[int] = None
    lte: Optional[int] = None
    in_: Optional[List[int]] = None
    isnull: Optional[bool] = None

# For dates
class DateTimeFilterInput:
    exact: Optional[str] = None
    gt: Optional[str] = None
    gte: Optional[str] = None
    lt: Optional[str] = None
    lte: Optional[str] = None
    date: Optional[str] = None  # Filter by date without time
    year: Optional[int] = None
    month: Optional[int] = None
    day: Optional[int] = None
    isnull: Optional[bool] = None

Usage in GraphQL:

query {
  users(filters: {
    name: { icontains: "john" }
    age: { gte: 18, lt: 65 }
    created_at: { date: "2024-01-01" }
    status: { in_: ["active", "pending"] }
  }) {
    items { id name }
  }
}

Pagination

Support for two pagination types:

# Limit/Offset pagination
class UserViewSet(AutoGraphQLViewSet):
    model = User
    pagination_class = LimitOffsetPagination(default_limit=20, max_limit=100)

# Page-based pagination
class PostViewSet(AutoGraphQLViewSet):
    model = Post
    pagination_class = PageNumberPagination(page_size=10, max_page_size=50)

# Limit/Offset
query {
  users(limit: 10, offset: 20) {
    items { id name }
    pagination {
      totalCount
      limit
      offset
      hasNext
      hasPrevious
    }
  }
}

# Page-based
query {
  posts(page: 2, pageSize: 15) {
    items { id title }
    pagination {
      totalCount
      page
      totalPages
      hasNext
      hasPrevious
    }
  }
}

Type Customization

Full control over type generation:

from models_to_graphql import run_gql_generation

# Type generation
models = {"User": User, "Post": Post}
result = run_gql_generation(models)

# Get components
types = result["types"]          # Strawberry types
queries = result["queries"]      # Query resolvers
mutations = result["mutations"]  # Mutation resolvers

# Type customization
@strawberry.type
class CustomUser:
    id: int
    name: str
    email: str
    
    @strawberry.field
    async def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"
    
    @strawberry.field
    async def posts_count(self, info) -> int:
        session = info.context["session"]
        return await session.scalar(
            select(func.count(Post.id)).where(Post.author_id == self.id)
        )

# Use custom type
class UserViewSet(AutoGraphQLViewSet):
    model = User
    graphql_type = CustomUser  # Override auto-generated type

Custom Resolvers

class UserViewSet(AutoGraphQLViewSet):
    model = User
    
    # Custom query resolver
    async def list(self, filter_params=None, limit=None, offset=0, ordering=None):
        # Can completely override logic
        async with self.get_session() as session:
            # Your custom logic
            stmt = select(User).where(User.is_verified == True)
            
            # Apply basic filtering
            if filter_params:
                stmt = await self.filter_queryset(stmt, filter_params, session)
            
            # Execute query
            result = await session.execute(stmt.limit(limit).offset(offset))
            items = result.scalars().all()
            
            # Convert to GraphQL types
            return [self._convert_model_to_graphql(item) for item in items]
    
    # Custom mutation resolver
    async def create(self, input_data):
        # Additional validation
        if not self.validate_email(input_data.email):
            raise ValueError("Invalid email format")
        
        # Call base logic
        return await super().create(input_data)

Apollo Federation

Microservice architecture support:

class UserViewSet(AutoGraphQLViewSet):
    model = User
    
    # Federation settings
    _federation_entity = True
    _federation_key_fields = ['id']
    _federation_shareable_fields = ['name', 'email']
    _federation_external_fields = ['organization_id']

# Export federated schema
from models_to_graphql.schema.export import export_schemas_from_viewsets

export_schemas_from_viewsets(
    viewsets_config=[("user", UserViewSet)],
    service_name="user_service",
    federation_config={
        "service_url": "http://localhost:8001/graphql",
        "enable_federation_2": True
    }
)

TypeScript Export

Automatic TypeScript code generation:

from models_to_graphql.schema.typescript_export import export_typescript

# TypeScript export
typescript_path = export_typescript(
    viewsets_config=[("user", UserViewSet), ("post", PostViewSet)],
    service_name="my_service"
)

# Creates structure:
# export/my_service/
# ├── types.ts        # TypeScript interfaces
# ├── queries.ts      # GraphQL queries  
# ├── mutations.ts    # GraphQL mutations
# └── index.ts        # Exports

Result in types.ts:

export interface User {
  id: string;
  name: string;
  email?: string;
  posts: Array<Post>;
}

export interface Post {
  id: string;
  title: string;
  content?: string;
  author?: User;
}

Result in queries.ts:

export const USERS_QUERY = gql`
  query UsersQuery($filters: UserFilter, $limit: Int, $offset: Int) {
    users(filters: $filters, limit: $limit, offset: $offset) {
      items {
        id
        name
        email
        posts {
          id
          title
        }
      }
      pagination {
        totalCount
        hasNext
        hasPrevious
      }
    }
  }
`;

🔄 Model Relationships

Automatic SQLAlchemy relationship resolution:

# SQLAlchemy models
class Author(Base):
    __tablename__ = "authors"
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    books = relationship("Book", back_populates="author")

class Book(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    author_id = Column(Integer, ForeignKey("authors.id"))
    author = relationship("Author", back_populates="books")

# ViewSets
class AuthorViewSet(AutoGraphQLViewSet):
    model = Author

class BookViewSet(AutoGraphQLViewSet):
    model = Book

# Create schema with relationships
schema = create_graphql_schema_from_viewsets([
    ("author", AuthorViewSet),
    ("book", BookViewSet),
], with_relationships=True)  # Enable relationships

GraphQL query with automatically resolved relationships:

query {
  authors {
    items {
      id
      name
      books {  # Automatically resolved relationship
        id
        title
      }
    }
  }
  
  books {
    items {
      id
      title
      author {  # Automatically resolved relationship
        id
        name
      }
    }
  }
}

🛠 Advanced Customization

Custom Input Types

import strawberry
from typing import Optional

@strawberry.input
class CreateUserInput:
    name: str
    email: str
    age: Optional[int] = None
    preferences: Optional[str] = None  # JSON string

@strawberry.input  
class UpdateUserInput:
    name: Optional[str] = None
    email: Optional[str] = None
    age: Optional[int] = None

class UserViewSet(AutoGraphQLViewSet):
    model = User
    
    # Override auto-generated input types
    create_input = CreateUserInput
    update_input = UpdateUserInput
    
    async def prepare_create_data(self, data, input_data):
        data = await super().prepare_create_data(data, input_data)
        
        # Handle JSON preferences
        if 'preferences' in data and data['preferences']:
            import json
            data['preferences'] = json.loads(data['preferences'])
            
        return data

Custom Filters

@strawberry.input
class UserFilter:
    name: Optional[StringFilterInput] = None
    email: Optional[StringFilterInput] = None
    age: Optional[IntFilterInput] = None
    is_premium: Optional[bool] = None  # Custom filter

class UserViewSet(AutoGraphQLViewSet):
    model = User
    filter_input = UserFilter  # Override auto-generated filter
    
    async def filter_queryset(self, queryset, filter_params, session):
        # Apply base filters
        queryset = await super().filter_queryset(queryset, filter_params, session)
        
        # Handle custom filter
        if hasattr(filter_params, 'is_premium') and filter_params.is_premium is not None:
            if filter_params.is_premium:
                queryset = queryset.where(User.subscription_type == 'premium')
            else:
                queryset = queryset.where(User.subscription_type != 'premium')
        
        return queryset

Middleware and Context

from typing import Any, Dict

class DatabaseMiddleware:
    async def __call__(self, resolver, obj, info, **kwargs):
        # Add session to context
        async with get_db_session() as session:
            info.context["session"] = session
            return await resolver(obj, info, **kwargs)

class AuthMiddleware:
    async def __call__(self, resolver, obj, info, **kwargs):
        # Check authorization
        token = info.context.get("request").headers.get("Authorization")
        user = await verify_token(token)
        info.context["current_user"] = user
        return await resolver(obj, info, **kwargs)

# Usage with FastAPI
from strawberry.fastapi import GraphQLRouter

graphql_app = GraphQLRouter(
    schema,
    context_getter=lambda request: {"request": request},
    middleware=[DatabaseMiddleware(), AuthMiddleware()]
)

Complex Business Logic

class OrderViewSet(AutoGraphQLViewSet):
    model = Order
    
    async def pre_create(self, session, input_data):
        # Check product availability
        product_ids = [item.product_id for item in input_data.items]
        products = await session.execute(
            select(Product).where(Product.id.in_(product_ids))
        )
        
        if len(products.scalars().all()) != len(product_ids):
            raise ValueError("Some products not found")
    
    async def post_create(self, obj, input_data):
        # Notify about new order
        await self.send_order_notification(obj)
        
        # Update inventory
        for item in input_data.items:
            await self.update_product_stock(item.product_id, -item.quantity)
        
        return obj
    
    async def pre_update(self, session, obj, input_data):
        # Check order status
        if obj.status == 'shipped':
            raise ValueError("Cannot modify shipped order")
    
    async def pre_delete(self, session, obj):
        # Check if order can be deleted
        if obj.status in ['shipped', 'delivered']:
            raise ValueError("Cannot delete shipped or delivered order")

Performance Optimization

class UserViewSet(AutoGraphQLViewSet):
    model = User
    
    # Enable query optimization
    select_related = ['profile', 'organization']  # Eager loading
    prefetch_related = ['posts', 'comments']      # Separate queries
    
    # Custom query optimization
    async def optimize_query(self, stmt, info):
        # Analyze GraphQL selection set
        fields = self.get_selected_fields(info)
        
        if 'posts' in fields:
            stmt = stmt.options(selectinload(User.posts))
        
        if 'organization' in fields:
            stmt = stmt.options(joinedload(User.organization))
        
        return stmt
    
    # Caching
    cache_timeout = 300  # 5 minutes
    
    async def get_cache_key(self, method, **kwargs):
        return f"user:{method}:{hash(str(kwargs))}"

Error Handling

from models_to_graphql.exceptions import ValidationError, NotFoundError

class UserViewSet(AutoGraphQLViewSet):
    model = User
    
    async def handle_error(self, error, operation, **kwargs):
        # Log error
        logger.error(f"Error in {operation}: {error}", exc_info=True)
        
        # Convert to GraphQL error
        if isinstance(error, IntegrityError):
            if 'email' in str(error):
                raise ValidationError("Email already exists")
            raise ValidationError("Data integrity error")
        
        if isinstance(error, NoResultFound):
            raise NotFoundError("User not found")
        
        # Let parent handle other errors
        return await super().handle_error(error, operation, **kwargs)

Testing

import pytest
from models_to_graphql.testing import GraphQLTestCase

class TestUserViewSet(GraphQLTestCase):
    viewset_class = UserViewSet
    
    async def test_create_user(self):
        mutation = """
        mutation {
            create_user(input: {
                name: "Test User"
                email: "test@example.com"
            }) {
                id
                name
                email
            }
        }
        """
        
        result = await self.execute_graphql(mutation)
        assert not result.errors
        assert result.data['create_user']['name'] == "Test User"
    
    async def test_list_users_with_filters(self):
        # Create test data
        await self.create_test_users()
        
        query = """
        query {
            users(filters: { name: { icontains: "john" } }) {
                items {
                    id
                    name
                }
            }
        }
        """
        
        result = await self.execute_graphql(query)
        assert len(result.data['users']['items']) == 2

🚀 Production Deployment

Docker Setup

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Environment Configuration

from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str
    redis_url: str = None
    debug: bool = False
    cors_origins: list = ["*"]
    
    # GraphQL settings
    graphql_introspection: bool = False
    graphql_playground: bool = False
    
    class Config:
        env_file = ".env"

settings = Settings()

Monitoring and Metrics

from models_to_graphql.middleware import MetricsMiddleware, TracingMiddleware

# Add monitoring middleware
graphql_app = GraphQLRouter(
    schema,
    middleware=[
        MetricsMiddleware(),  # Prometheus metrics
        TracingMiddleware(), # OpenTelemetry tracing
        DatabaseMiddleware(),
        AuthMiddleware()
    ]
)

📊 Examples and Use Cases

E-commerce API

# Models
class Category(Base):
    __tablename__ = "categories"
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    products = relationship("Product", back_populates="category")

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String(200))
    price = Column(Decimal(10, 2))
    category_id = Column(Integer, ForeignKey("categories.id"))
    category = relationship("Category", back_populates="products")

# ViewSets
class ProductViewSet(AutoGraphQLViewSet):
    model = Product
    
    async def filter_queryset(self, queryset, filter_params, session):
        queryset = await super().filter_queryset(queryset, filter_params, session)
        # Only show available products
        return queryset.where(Product.is_available == True)

# Usage
schema = create_graphql_schema_from_viewsets([
    ("category", CategoryViewSet),
    ("product", ProductViewSet),
], with_relationships=True)

Blog API

class BlogViewSet(AutoGraphQLViewSet):
    model = Post
    
    # Custom ordering
    default_ordering = ['-created_at']
    
    # Search functionality
    search_fields = ['title', 'content']
    
    async def list(self, search=None, **kwargs):
        if search:
            # Implement full-text search
            kwargs['filters'] = {
                'title': {'icontains': search}
            }
        return await super().list(**kwargs)

Multi-tenant SaaS

class TenantAwareViewSet(AutoGraphQLViewSet):
    
    async def filter_queryset(self, queryset, filter_params, session):
        queryset = await super().filter_queryset(queryset, filter_params, session)
        
        # Add tenant filtering
        current_user = self.get_current_user()
        return queryset.where(self.model.tenant_id == current_user.tenant_id)

class UserViewSet(TenantAwareViewSet):
    model = User

class ProjectViewSet(TenantAwareViewSet):
    model = Project

🤝 Contributing

We welcome contributions! Please feel free to submit issues and pull requests.

Development Setup

git clone https://github.com/rostislav444/models-to-graphql.git
cd models-to-graphql

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov=models_to_graphql

# Format code
black .
isort .

# Type checking
mypy models_to_graphql

📝 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

SQLAlchemy - The Python SQL Toolkit
Strawberry GraphQL - GraphQL library for Python
Django REST Framework - Inspiration for ViewSets
FastAPI - Modern web framework for Python

Name		Name	Last commit message	Last commit date
Latest commit History 3 Commits
examples		examples
models_to_graphql		models_to_graphql
.gitignore		.gitignore
LICENCE		LICENCE
MANIFEST.in		MANIFEST.in
cli.py		cli.py
pyproject.toml		pyproject.toml
readme.md		readme.md
requirements.txt		requirements.txt
setup.py		setup.py

Folders and files

Latest commit

History

Repository files navigation

Models to GraphQL

✨ Features

🔧 Installation

🚀 Quick Start

1. Define SQLAlchemy models

2. Create ViewSets

3. Create GraphQL schema

4. Done! 🎉

📚 Detailed Documentation

ViewSets - Architecture Foundation

Typed Filtering

Pagination

Type Customization

Custom Resolvers

Apollo Federation

TypeScript Export

🔄 Model Relationships

🛠 Advanced Customization

Custom Input Types

Custom Filters

Middleware and Context

Complex Business Logic

Performance Optimization

Error Handling

Testing

🚀 Production Deployment

Docker Setup

Environment Configuration

Monitoring and Metrics

📊 Examples and Use Cases

E-commerce API

Blog API

Multi-tenant SaaS

🤝 Contributing

Development Setup

📝 License

🙏 Acknowledgments

About

Topics

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages