Library

Pony ORM

Overview

Pony ORM is an innovative Python ORM library designed as a "unique ORM that allows writing SQL queries in Pythonic syntax". Its greatest feature is the ability to write database queries intuitively using Python generator expressions and lambda functions. With an approach that sets it apart from traditional ORMs, you can write complex queries with just Python knowledge without knowing SQL, earning high praise in educational settings and small-to-medium scale projects.

Details

Pony ORM 2025 version receives passionate support from some developers due to its special syntax. By leveraging Python generator expressions, SQL queries can be written in forms close to natural language. It adopts the Data Mapper pattern, making the relationship between Python objects and database tables transparent. Debug mode allows you to check generated SQL, providing high educational value. Its use for educational purposes is increasing, being appreciated for its ability to deepen Python understanding while learning database operations.

Key Features

• Pythonic Queries: Natural query writing using generator expressions
• Intuitive Syntax: Query creation with Python only, no SQL knowledge required
• Automatic Optimization: Automatic query optimization and SQL generation
• Debug Support: Generated SQL verification and query analysis features
• Type Safety: Natural integration with Python's type system
• Lightweight Design: High functionality with minimal configuration

Pros and Cons

Pros

• Python beginners can perform database operations without SQL knowledge
• Generator expressions enable extremely readable query code
• Optimal for educational purposes, deepening understanding of Python and databases
• Rich debugging features allow checking generated SQL
• High development efficiency in small-to-medium scale projects
• Unique approach stimulates developer learning motivation

Cons

• Special syntax creates learning costs in team development
• May not be suitable for large-scale, complex applications
• Difficult to use alongside other ORMs or migrate from them
• Limited adoption cases at enterprise level
• Constrained performance tuning options
• Small community and plugin ecosystem

Reference Pages

• Pony ORM Official Site
• Pony ORM Documentation
• Pony ORM GitHub Repository

Code Examples

Installation and Setup

# Install Pony ORM
pip install pony

# Database-specific adapters
pip install pony[mysql]      # For MySQL
pip install pony[postgresql] # For PostgreSQL

# SQLite is included by default

Basic Model Definition and Database Configuration

from pony.orm import *
from datetime import datetime
import os

# Database connection
db = Database()

# Entity (model) definition
class Customer(db.Entity):
    id = PrimaryKey(int, auto=True)
    name = Required(str)
    email = Required(str, unique=True)
    country = Optional(str)
    city = Optional(str)
    orders = Set('Order')

class Product(db.Entity):
    id = PrimaryKey(int, auto=True)
    name = Required(str)
    price = Required(float)
    description = Optional(str)
    category = Required('Category')
    order_items = Set('OrderItem')

class Category(db.Entity):
    id = PrimaryKey(int, auto=True)
    name = Required(str, unique=True)
    products = Set(Product)

class Order(db.Entity):
    id = PrimaryKey(int, auto=True)
    date = Required(datetime, default=datetime.now)
    customer = Required(Customer)
    total_price = Required(float)
    items = Set('OrderItem')

class OrderItem(db.Entity):
    order = Required(Order)
    product = Required(Product)
    quantity = Required(int)
    price = Required(float)
    PrimaryKey(order, product)

# Database binding and configuration
# SQLite (for development/testing)
db.bind('sqlite', 'example.db')

# PostgreSQL (for production)
# db.bind('postgres', user='username', password='password', 
#         host='localhost', database='mydb')

# MySQL (for production)
# db.bind('mysql', host='localhost', user='username', password='password', 
#         db='mydb')

# Enable debug mode (SQL output)
set_sql_debug(True)

# Generate schema
db.generate_mapping(create_tables=True)

# Populate sample data
@db_session
def populate_sample_data():
    if Category.select().count() == 0:
        # Create categories
        electronics = Category(name="Electronics")
        books = Category(name="Books") 
        clothes = Category(name="Clothes")
        
        # Create products
        Product(name="Laptop", price=999.99, category=electronics)
        Product(name="Mouse", price=25.50, category=electronics)
        Product(name="Python Book", price=45.00, category=books)
        Product(name="T-Shirt", price=19.99, category=clothes)
        
        # Create customers
        Customer(name="John Smith", email="[email protected]", country="USA", city="New York")
        Customer(name="Alice Johnson", email="[email protected]", country="Canada", city="Toronto")
        
        commit()

populate_sample_data()

Basic Pythonic Queries

@db_session
def basic_queries():
    # Basic select (generator expression)
    customers = select(c for c in Customer)
    print("All customers:", list(customers))
    
    # Conditional select
    usa_customers = select(c for c in Customer if c.country == "USA")
    print("USA customers:", list(usa_customers))
    
    # Complex conditions
    expensive_electronics = select(p for p in Product 
                                  if p.category.name == "Electronics" and p.price > 100)
    print("Expensive electronics:", list(expensive_electronics))
    
    # Using lambda expressions (entity methods)
    customers_lambda = Customer.select(lambda c: c.country == "USA")
    print("USA customers (lambda):", list(customers_lambda))
    
    # Single record retrieval
    customer = Customer.get(name="John Smith")
    print(f"Customer: {customer.name} from {customer.city}")
    
    # Existence check
    exists = Customer.exists(name="Alice Johnson")
    print(f"Alice exists: {exists}")
    
    # Count
    total_customers = Customer.select().count()
    print(f"Total customers: {total_customers}")

Advanced Queries and Joins

@db_session
def advanced_queries():
    # JOIN operations (automatic)
    products_with_category = select(p for p in Product)
    for p in products_with_category:
        print(f"{p.name} - Category: {p.category.name}")
    
    # Complex conditions with JOIN
    customers_with_orders = select(c for c in Customer 
                                  if count(c.orders) > 0)
    print("Customers with orders:", list(customers_with_orders))
    
    # Nested queries
    popular_products = select(p for p in Product 
                             if count(p.order_items) > 0)
    print("Popular products:", list(popular_products))
    
    # Aggregate functions
    total_sales = sum(o.total_price for o in Order)
    print(f"Total sales: ${total_sales}")
    
    # Group aggregation
    sales_by_country = select((c.country, sum(o.total_price)) 
                             for c in Customer 
                             for o in c.orders 
                             if c.country is not None)
    print("Sales by country:", list(sales_by_country))
    
    # Maximum and minimum values
    max_price = max(p.price for p in Product)
    min_price = min(p.price for p in Product)
    print(f"Price range: ${min_price} - ${max_price}")
    
    # Average values
    avg_order_value = avg(o.total_price for o in Order)
    print(f"Average order value: ${avg_order_value}")

Complex Queries and Subqueries

@db_session
def complex_queries():
    # Subqueries
    expensive_product_customers = select(c for c in Customer 
                                        for o in c.orders 
                                        for item in o.items 
                                        if item.product.price > 500)
    print("Customers who bought expensive products:", 
          list(expensive_product_customers))
    
    # EXISTS equivalent processing
    customers_without_orders = select(c for c in Customer 
                                     if not c.orders)
    print("Customers without orders:", list(customers_without_orders))
    
    # IN operator equivalent
    electronic_categories = select(c for c in Category 
                                  if c.name in ["Electronics", "Computers"])
    print("Electronic categories:", list(electronic_categories))
    
    # String operations
    customers_with_gmail = select(c for c in Customer 
                                 if "gmail" in c.email)
    print("Gmail customers:", list(customers_with_gmail))
    
    # Date range
    from datetime import datetime, timedelta
    recent_orders = select(o for o in Order 
                          if o.date > datetime.now() - timedelta(days=30))
    print("Recent orders:", list(recent_orders))
    
    # Complex aggregation
    customer_order_stats = select((c.name, 
                                  count(c.orders), 
                                  sum(c.orders.total_price),
                                  avg(c.orders.total_price))
                                 for c in Customer 
                                 if count(c.orders) > 0)
    print("Customer order statistics:")
    for name, order_count, total, average in customer_order_stats:
        print(f"  {name}: {order_count} orders, ${total:.2f} total, ${average:.2f} avg")

Data Insert, Update, Delete

@db_session
def crud_operations():
    # Insert
    new_customer = Customer(
        name="Bob Wilson",
        email="[email protected]",
        country="UK",
        city="London"
    )
    
    # Create related objects simultaneously
    electronics = Category.get(name="Electronics")
    new_product = Product(
        name="Smartphone",
        price=699.99,
        description="Latest smartphone",
        category=electronics
    )
    
    # Commit (automatically committed with @db_session)
    print(f"Created customer: {new_customer.name}")
    print(f"Created product: {new_product.name}")
    
    # Update
    customer_to_update = Customer.get(name="Bob Wilson")
    customer_to_update.city = "Manchester"
    print(f"Updated customer city: {customer_to_update.city}")
    
    # Bulk update
    update(c.country for c in Customer if c.country == "USA").set("United States")
    
    # Delete
    customer_to_delete = Customer.get(name="Bob Wilson")
    customer_to_delete.delete()
    print("Customer deleted")
    
    # Bulk delete
    delete(p for p in Product if p.price < 20)
    print("Deleted low-price products")

Transaction Management

# Using decorator
@db_session
def simple_transaction():
    # This entire function is one transaction
    customer = Customer(name="Transaction Test", email="[email protected]")
    product = Product(name="Test Product", price=100, category=Category.get(name="Electronics"))
    # Auto-commit

# Explicit transaction control
def manual_transaction():
    with db_session:
        try:
            customer = Customer(name="Manual Transaction", email="[email protected]")
            # Some business logic
            if some_condition_fails():
                rollback()  # Explicit rollback
            else:
                commit()   # Explicit commit
        except Exception as e:
            rollback()
            print(f"Transaction failed: {e}")

# Nested transactions
@db_session
def nested_transactions():
    customer = Customer(name="Outer Transaction", email="[email protected]")
    
    try:
        with db_session:
            # Inner transaction
            product = Product(name="Inner Product", price=50, 
                            category=Category.get(name="Electronics"))
            if some_validation_fails():
                raise ValueError("Validation failed")
    except ValueError:
        print("Inner transaction failed, but outer continues")
    
    # Outer transaction continues
    commit()

Raw SQL Execution and raw_sql Functions

@db_session
def raw_sql_examples():
    # Raw SQL select
    results = db.select("SELECT name, email FROM Customer WHERE country = $country", 
                       {"country": "USA"})
    print("Raw SQL results:", results)
    
    # Raw SQL with variables
    country = "Canada"
    canadian_customers = db.select("SELECT * FROM Customer WHERE country = $country", 
                                  {"country": country})
    print("Canadian customers:", canadian_customers)
    
    # Using raw_sql function in queries
    from pony.orm import raw_sql
    
    # Using date functions
    recent_orders_raw = select(o for o in Order 
                              if raw_sql("DATE(o.date)") == raw_sql("DATE('now')"))
    print("Today's orders:", list(recent_orders_raw))
    
    # Database-specific functions
    # PostgreSQL example
    # substring_customers = select(c for c in Customer 
    #                             if raw_sql("SUBSTRING(c.name, 1, 1)") == "J")
    
    # SQLite example  
    first_letter_j = select(c for c in Customer 
                           if raw_sql("SUBSTR(c.name, 1, 1)") == "J")
    print("Customers starting with J:", list(first_letter_j))

Performance Optimization

@db_session
def performance_optimization():
    # Prefetching (solving N+1 problem)
    # Bad: N+1 queries
    products = Product.select()
    for p in products:
        print(f"{p.name} - {p.category.name}")  # Queries category for each product
    
    # Good: Using JOIN
    products_with_category = select(p for p in Product)
    for p in products_with_category:
        print(f"{p.name} - {p.category.name}")  # Single JOIN query
    
    # Batch processing
    batch_size = 100
    total_products = Product.select().count()
    
    for offset in range(0, total_products, batch_size):
        batch = Product.select().limit(batch_size, offset=offset)
        process_product_batch(batch)
    
    # Index utilization
    # Index-efficient search
    indexed_customer = Customer.get(email="[email protected]")  # email has index
    
    # Index hints
    customers_by_country = select(c for c in Customer 
                                 if c.country == "USA").order_by(Customer.name)
    
    # Aggregation optimization
    # Efficient aggregation
    category_product_counts = select((c.name, count(c.products)) 
                                   for c in Category)
    print("Products per category:", list(category_product_counts))

def process_product_batch(products):
    for product in products:
        # Batch processing logic
        pass

# Query result caching
@db_session
def caching_example():
    # Convert results to list for caching
    expensive_products = list(select(p for p in Product if p.price > 100))
    
    # Reuse later
    for product in expensive_products:
        print(product.name)
    
    # Cache count()
    total_count = Product.select().count()
    print(f"Total products: {total_count}")

Debugging and Logging

from pony.orm import sql_debug

# Enable/disable SQL debugging
sql_debug(True)   # Start SQL output
sql_debug(False)  # Stop SQL output

@db_session
def debug_queries():
    # Check SQL before and after query execution
    print("=== Debug Query Start ===")
    
    customers = select(c for c in Customer if c.country == "USA")
    print("Query object created")
    
    # SQL is executed at this point
    customer_list = list(customers)
    print(f"Found {len(customer_list)} customers")
    
    print("=== Debug Query End ===")

# Custom log configuration
import logging

# PonyORM log configuration
logging.basicConfig(level=logging.DEBUG)
pony_logger = logging.getLogger('pony.orm')
pony_logger.setLevel(logging.DEBUG)

# Add handler
handler = logging.FileHandler('pony_queries.log')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
pony_logger.addHandler(handler)

# Query execution time measurement
@db_session
def measure_query_performance():
    import time
    
    start_time = time.time()
    
    # Complex query
    result = select((c.name, count(c.orders), sum(c.orders.total_price))
                   for c in Customer 
                   if count(c.orders) > 0)
    
    query_result = list(result)
    
    end_time = time.time()
    execution_time = end_time - start_time
    
    print(f"Query executed in {execution_time:.4f} seconds")
    print(f"Results: {len(query_result)} rows")