Overview
EntityManager is the core component for managing database operations in Framefox. It handles session management, transactions, entity persistence, and provides an identity map pattern for efficient entity tracking.
Class Definition
from framefox.core.orm import EntityManager
class EntityManager:
def __init__(self, connection_name: str = "default"):
...
Constructor
init
Initializes an EntityManager with a specific database connection.
def __init__(self, connection_name: str = "default")
The name of the database connection to use
Example
from framefox.core.orm import EntityManager
# Use default connection
em = EntityManager()
# Use a named connection
em_analytics = EntityManager(connection_name="analytics")
Properties
session
Returns the active database session or creates a new one if none exists.
@property
def session(self) -> Session
The active SQLModel session instance
Example
em = EntityManager()
session = em.session
# Use session directly for raw queries
result = session.exec(select(User)).all()
Session Management
close_session
Closes the active session if it exists.
def close_session(self) -> None
Example
em = EntityManager()
# ... perform operations ...
em.close_session()
transaction
Context manager for handling database transactions with automatic commit/rollback.
@contextlib.contextmanager
def transaction(self) -> Generator[Session, None, None]
return
Generator[Session, None, None]
A context manager yielding the session
Example
em = EntityManager()
try:
with em.transaction() as session:
user = User(username="john_doe", email="john@example.com")
session.add(user)
# Transaction commits automatically if no exception
except Exception as e:
# Transaction rolls back automatically on exception
print(f"Transaction failed: {e}")
commit
Commits the current transaction if not in a nested transaction.
Example
em = EntityManager()
user = User(username="john_doe", email="john@example.com")
em.persist(user)
em.commit()
rollback
Rolls back the current transaction.
def rollback(self) -> None
Example
em = EntityManager()
try:
user = User(username="john_doe", email="john@example.com")
em.persist(user)
# Something goes wrong
raise ValueError("Invalid data")
except ValueError:
em.rollback()
Entity Operations
persist
Adds an entity to the session and identity map. Handles entity relationships and automatically hydrates foreign key references.
def persist(self, entity) -> Any
The entity instance to persist
The persisted entity (may be merged if already attached to another session)
Example
em = EntityManager()
# Create and persist a new user
user = User(username="john_doe", email="john@example.com")
em.persist(user)
em.commit()
print(f"User ID: {user.id}") # ID assigned after commit
delete
Deletes the specified entity from the database.
def delete(self, entity) -> None
The entity instance to delete
Example
em = EntityManager()
user = em.find(User, 123)
if user:
em.delete(user)
em.commit()
print("User deleted")
refresh
Refreshes the entity’s state from the database, overwriting any local changes.
def refresh(self, entity) -> None
The entity instance to refresh
Example
em = EntityManager()
user = em.find(User, 123)
# Make local changes
user.username = "new_username"
# Discard changes and reload from database
em.refresh(user)
print(user.username) # Original value from database
Query Operations
exec_statement
Executes a SQL statement and returns the results as a list.
def exec_statement(self, statement) -> list
statement
Select | Delete | Update
required
The SQLModel statement to execute
List of results from the query
Example
from sqlmodel import select
em = EntityManager()
statement = select(User).where(User.username == "john_doe")
users = em.exec_statement(statement)
for user in users:
print(user.email)
find
Retrieves an entity by its primary key, using the identity map cache when possible.
def find(self, entity_class, primary_key) -> Any
The entity class to query
The entity instance, or None if not found
Example
em = EntityManager()
# Find by single primary key
user = em.find(User, 123)
# Find by composite primary key
order_item = em.find(OrderItem, {"order_id": 1, "product_id": 5})
find_existing_entity
Finds an existing entity in the database using the entity’s primary keys.
def find_existing_entity(self, entity) -> Any
The entity instance with primary keys set
The existing entity from the database, or None if not found
Example
em = EntityManager()
# Create entity with known ID
user = User(id=123)
# Check if it exists in database
existing_user = em.find_existing_entity(user)
if existing_user:
print(f"Found: {existing_user.username}")
Schema Management
create_all_tables
Creates all tables defined in the SQLModel metadata.
def create_all_tables(self) -> None
Example
em = EntityManager()
em.create_all_tables()
print("All tables created")
drop_all_tables
Drops all tables defined in the SQLModel metadata.
def drop_all_tables(self) -> None
This operation is destructive and will delete all data. Use with caution, typically only in development or testing.
Example
em = EntityManager()
em.drop_all_tables()
print("All tables dropped")
Repository Access
get_repository
Retrieves the repository instance associated with a given entity class.
def get_repository(self, entity_class: Type) -> Any
The entity class to find the repository for
The repository instance, or None if not found
Example
em = EntityManager()
user_repo = em.get_repository(User)
if user_repo:
all_users = user_repo.find_all()
Complete Example
from framefox.core.orm import EntityManager, AbstractEntity
from sqlmodel import Field, select
class User(AbstractEntity, table=True):
id: int | None = Field(default=None, primary_key=True)
username: str
email: str
is_active: bool = True
class Post(AbstractEntity, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str
content: str
user_id: int = Field(foreign_key="user.id")
# Initialize EntityManager
em = EntityManager()
# Create tables
em.create_all_tables()
# Create entities with transaction
try:
with em.transaction():
# Create user
user = User(username="john_doe", email="john@example.com")
em.persist(user)
# Create post for user
post = Post(
title="My First Post",
content="Hello, World!",
user_id=user.id
)
em.persist(post)
# Transaction commits automatically
except Exception as e:
print(f"Transaction failed: {e}")
# Find entities
user = em.find(User, 1)
if user:
print(f"Found user: {user.username}")
# Execute custom query
statement = select(User).where(User.is_active == True)
active_users = em.exec_statement(statement)
for user in active_users:
print(user.email)
# Update entity
user.email = "newemail@example.com"
em.persist(user)
em.commit()
# Refresh entity from database
em.refresh(user)
# Delete entity
em.delete(user)
em.commit()
# Clean up
em.close_session()
Notes
- EntityManager is typically request-scoped in web applications
- The identity map pattern prevents loading the same entity multiple times
- Nested transactions are supported through transaction depth tracking
- Always close sessions when done to free database connections
- Use
transaction() context manager for automatic commit/rollback handling
- The
persist() method automatically hydrates relationships when foreign keys are provided