SQLAlchemy Integration
factorio works seamlessly with SQLAlchemy models. Since factorio doesn't interact with databases, you maintain full control over when and how objects are persisted.
Basic Setup
Define Your Models
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, create_engine
from sqlalchemy.orm import DeclarativeBase, relationship
from datetime import datetime
Base = DeclarativeBase()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, nullable=False)
age = Column(Integer)
created_at = Column(DateTime, default=datetime.utcnow)
# Relationship
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(String)
author_id = Column(Integer, ForeignKey('users.id'), nullable=False)
# Relationship
author = relationship("User", back_populates="posts")
Create Factories
from factorio import fields
from factorio.factories import Factory
class UserFactory(Factory[User]):
name = fields.TextField("name")
email = fields.TextField("email")
age = fields.IntegerField(min_value=18, max_value=65)
# Don't set created_at - let SQLAlchemy handle defaults
class PostFactory(Factory[Post]):
title = fields.TextField("sentence", nb_words=6)
content = fields.TextField("paragraphs", nb=3)
author = fields.FactoryField(UserFactory)
Usage in Tests
Creating Objects Without Persistence
def test_user_creation():
# Build object in memory (no database interaction)
user = UserFactory.build()
assert isinstance(user, User)
assert user.name is not None
assert "@" in user.email
assert 18 <= user.age <= 65
Persisting to Database
from sqlalchemy.orm import Session
def test_save_user_to_database(db_session: Session):
# Build object
user = UserFactory.build()
# Explicitly save to database
db_session.add(user)
db_session.commit()
# Verify it was saved
assert user.id is not None
saved_user = db_session.query(User).filter_by(email=user.email).first()
assert saved_user is not None
assert saved_user.name == user.name
Working with Relationships
def test_post_with_author(db_session: Session):
# Build post (automatically builds author too)
post = PostFactory.build()
assert isinstance(post.author, User)
assert post.author.name is not None
# Save both to database
db_session.add(post)
db_session.commit()
# Verify relationship
assert post.author_id is not None
assert post.author.id == post.author_id
Advanced Patterns
Overriding Foreign Keys
When you need specific relationships:
def test_post_with_existing_author(db_session: Session):
# Create and save an author first
existing_user = UserFactory.build()
db_session.add(existing_user)
db_session.commit()
# Create post with existing author
post = PostFactory.build(author=existing_user)
db_session.add(post)
db_session.commit()
assert post.author_id == existing_user.id
Bulk Creation
def test_create_multiple_users(db_session: Session):
# Create multiple users
users = [UserFactory.build() for _ in range(10)]
# Bulk insert
db_session.add_all(users)
db_session.commit()
assert len(users) == 10
assert all(user.id is not None for user in users)
Custom Field Ranges per Test
def test_adult_users(db_session: Session):
# Override age range for this specific test
adult_user = UserFactory.build(
age=fields.IntegerField(min_value=21, max_value=65)
)
db_session.add(adult_user)
db_session.commit()
assert adult_user.age >= 21
def test_teen_users(db_session: Session):
# Different age range
teen_user = UserFactory.build(
age=fields.IntegerField(min_value=18, max_value=20)
)
db_session.add(teen_user)
db_session.commit()
assert 18 <= teen_user.age <= 20
Testing Best Practices
Use Fixtures for Database Sessions
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
@pytest.fixture
def db_session():
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
yield session
session.rollback()
session.close()
Clean State Between Tests
@pytest.fixture(autouse=True)
def clean_db(db_session):
"""Automatically rollback after each test."""
yield
db_session.rollback()
Factory Helpers for Common Patterns
class UserFactory(Factory[User]):
name = fields.TextField("name")
email = fields.TextField("email")
age = fields.IntegerField(min_value=18, max_value=65)
@classmethod
def create_and_save(cls, session: Session, **kwargs) -> User:
"""Helper to build and save in one step."""
user = cls.build(**kwargs)
session.add(user)
session.commit()
return user
# Usage in tests
def test_quick_user_creation(db_session: Session):
user = UserFactory.create_and_save(db_session, age=25)
assert user.id is not None
assert user.age == 25
Common Pitfalls
⚠️ Don't Set Primary Keys in Factories
# ❌ Wrong - let the database assign IDs
class UserFactory(Factory[User]):
id = fields.IntegerField() # Don't do this!
name = fields.TextField("name")
# ✅ Correct - only set non-PK fields
class UserFactory(Factory[User]):
name = fields.TextField("name")
email = fields.TextField("email")
⚠️ Handle Unique Constraints
# ❌ May cause integrity errors
user1 = UserFactory.build()
user2 = UserFactory.build()
# Both might have same email!
# ✅ Override to ensure uniqueness
user1 = UserFactory.build(email="user1@example.com")
user2 = UserFactory.build(email="user2@example.com")
# Or use sequences (see Cookbook)
⚠️ Don't Forget to Commit
# ❌ Object won't be persisted
user = UserFactory.build()
db_session.add(user)
# Forgot to commit!
# ✅ Always commit after adding
user = UserFactory.build()
db_session.add(user)
db_session.commit()
Complete Example
import pytest
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from factorio import fields
from factorio.factories import Factory
# Models
Base = DeclarativeBase()
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
price = Column(Integer) # Price in cents
category = Column(String(50))
# Factory
class ProductFactory(Factory[Product]):
name = fields.TextField("word").capitalize()
price = fields.IntegerField(min_value=100, max_value=99999)
category = fields.ChoiceField(["electronics", "books", "clothing"])
# Test
@pytest.fixture
def db_session():
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
yield session
session.close()
def test_product_workflow(db_session):
# Build product
product = ProductFactory.build()
assert isinstance(product, Product)
assert product.name
assert 100 <= product.price <= 99999
# Save to database
db_session.add(product)
db_session.commit()
assert product.id is not None
# Query back
saved = db_session.query(Product).get(product.id)
assert saved.name == product.name
assert saved.price == product.price