Engines

The Engine is the starting point for any SQLAlchemy application. It manages database connections, connection pooling, and SQL dialect configuration.

Creating Engines

Basic Engine Creation

from sqlalchemy import create_engine

# SQLite in-memory
engine = create_engine('sqlite:///:memory:')

# SQLite file
engine = create_engine('sqlite:///database.db')

# PostgreSQL
engine = create_engine('postgresql://user:password@localhost/dbname')

# MySQL
engine = create_engine('mysql+pymysql://user:password@localhost/dbname')

# SQL Server
engine = create_engine('mssql+pyodbc://user:password@localhost/dbname')

create_engine

function

Signature: create_engine(url: str | URL, **kwargs) -> EngineCreates a new Engine instance. The first argument is a database URL, followed by optional configuration parameters.

URL Format

dialect[+driver]://username:password@host:port/database[?key=value]

Components:

dialect: Database type (postgresql, mysql, sqlite, mssql, oracle)
driver: DBAPI driver (psycopg2, pymysql, pyodbc, cx_oracle)
username: Database user
password: User password (URL-encoded)
host: Server hostname or IP
port: Server port (optional, uses default)
database: Database name
key=value: Query parameters

Database URLs

PostgreSQL

psycopg2
asyncpg
pg8000

# psycopg2 (most common)
engine = create_engine(
    'postgresql://scott:tiger@localhost/mydatabase'
)

# Explicit driver
engine = create_engine(
    'postgresql+psycopg2://scott:tiger@localhost/mydatabase'
)

# With port
engine = create_engine(
    'postgresql://scott:tiger@localhost:5432/mydatabase'
)

# SSL mode
engine = create_engine(
    'postgresql://scott:tiger@localhost/mydatabase?sslmode=require'
)

# Async PostgreSQL
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    'postgresql+asyncpg://scott:tiger@localhost/mydatabase'
)

# Pure Python driver
engine = create_engine(
    'postgresql+pg8000://scott:tiger@localhost/mydatabase'
)

MySQL / MariaDB

PyMySQL
mysqlclient
aiomysql

# PyMySQL (pure Python)
engine = create_engine(
    'mysql+pymysql://scott:tiger@localhost/foo'
)

# With charset
engine = create_engine(
    'mysql+pymysql://scott:tiger@localhost/foo?charset=utf8mb4'
)

# mysqlclient (C extension)
engine = create_engine(
    'mysql+mysqldb://scott:tiger@localhost/foo'
)

# Async MySQL
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    'mysql+aiomysql://scott:tiger@localhost/foo'
)

SQLite

# In-memory database
engine = create_engine('sqlite:///:memory:')

# File database (relative path)
engine = create_engine('sqlite:///foo.db')

# File database (absolute path)
engine = create_engine('sqlite:////absolute/path/to/foo.db')

# Windows absolute path
engine = create_engine('sqlite:///C:/path/to/foo.db')

SQL Server

# Windows Authentication
engine = create_engine(
    'mssql+pyodbc://hostname/dbname?driver=ODBC+Driver+17+for+SQL+Server'
)

# SQL Server Authentication
engine = create_engine(
    'mssql+pyodbc://scott:tiger@hostname/dbname?driver=ODBC+Driver+17+for+SQL+Server'
)

# Named instance
engine = create_engine(
    'mssql+pyodbc://scott:tiger@hostname\\instancename/dbname?driver=ODBC+Driver+17+for+SQL+Server'
)

Oracle

# cx_Oracle
engine = create_engine(
    'oracle+cx_oracle://scott:tiger@localhost:1521/?service_name=orcl'
)

# TNS name
engine = create_engine(
    'oracle+cx_oracle://scott:tiger@tnsname'
)

URL Object

Programmatic URL Construction

from sqlalchemy import URL

url = URL.create(
    drivername="postgresql+psycopg2",
    username="scott",
    password="tiger",
    host="localhost",
    port=5432,
    database="test",
    query={"sslmode": "require"}
)

engine = create_engine(url)

Password Encoding

from urllib.parse import quote_plus

# Password with special characters
password = "p@ssw0rd!#"
encoded = quote_plus(password)

url = f"postgresql://user:{encoded}@localhost/db"
engine = create_engine(url)

# Or use URL object
url = URL.create(
    "postgresql",
    username="user",
    password="p@ssw0rd!#",  # Automatically encoded
    host="localhost",
    database="db"
)

Parsing URLs

from sqlalchemy import make_url

url = make_url("postgresql://scott:tiger@localhost/test")

print(url.drivername)  # postgresql
print(url.username)    # scott
print(url.password)    # tiger
print(url.host)        # localhost
print(url.database)    # test

Engine Configuration

Common Parameters

engine = create_engine(
    'postgresql://scott:tiger@localhost/test',
    
    # Echo SQL statements
    echo=True,
    
    # Echo results
    echo_pool=False,
    
    # Connection pool size
    pool_size=5,
    
    # Max overflow connections
    max_overflow=10,
    
    # Pool timeout
    pool_timeout=30,
    
    # Connection recycle time (seconds)
    pool_recycle=3600,
    
    # Pre-ping connections
    pool_pre_ping=True,
    
    # Use LIFO for pool
    pool_use_lifo=False,
    
    # Execution options
    execution_options={
        "isolation_level": "REPEATABLE READ"
    }
)

Echo and Logging

# Print all SQL statements
engine = create_engine(url, echo=True)

# Output:
# SELECT users.id, users.name FROM users WHERE users.id = ?
# [1]

Pool Configuration

# Custom pool settings
engine = create_engine(
    url,
    pool_size=20,           # Keep 20 connections
    max_overflow=30,        # Allow 30 additional connections
    pool_timeout=60,        # Wait 60s for connection
    pool_recycle=1800,      # Recycle connections after 30 min
    pool_pre_ping=True,     # Test connections before use
)

pool_pre_ping=True is recommended for applications with idle connections. It prevents “connection has gone away” errors.

Dialect-Specific Options

engine = create_engine(
    'postgresql://scott:tiger@localhost/test',
    
    # Use server-side cursors
    server_side_cursors=True,
    
    # JSON serializer/deserializer
    json_serializer=json.dumps,
    json_deserializer=json.loads,
    
    # Client encoding
    client_encoding='utf8',
    
    # Connect args
    connect_args={
        'application_name': 'myapp',
        'options': '-c timezone=utc'
    }
)

Connection Testing

# Test connection
try:
    with engine.connect() as conn:
        result = conn.execute(text("SELECT 1"))
        print("Connection successful!")
except Exception as e:
    print(f"Connection failed: {e}")

# Get raw DBAPI connection
with engine.connect() as conn:
    dbapi_conn = conn.connection.dbapi_connection
    print(f"Connected to: {dbapi_conn}")

Engine Disposal

# Close all connections
engine.dispose()

# Recommended for long-running applications
# that need to recreate the pool

# After dispose, new connections will be created
with engine.connect() as conn:
    # Fresh connection from new pool
    result = conn.execute(text("SELECT 1"))

Engine Events

Connection Events

from sqlalchemy import event

@event.listens_for(engine, "connect")
def receive_connect(dbapi_connection, connection_record):
    """Called when a new connection is created."""
    print("New connection established")

@event.listens_for(engine, "checkout")
def receive_checkout(dbapi_connection, connection_record, connection_proxy):
    """Called when a connection is retrieved from the pool."""
    print("Connection checked out")

@event.listens_for(engine, "checkin")
def receive_checkin(dbapi_connection, connection_record):
    """Called when a connection is returned to the pool."""
    print("Connection checked in")

Execution Events

@event.listens_for(engine, "before_cursor_execute")
def receive_before_cursor_execute(
    conn, cursor, statement, parameters, context, executemany
):
    """Intercept statement before execution."""
    print(f"Executing: {statement}")

@event.listens_for(engine, "after_cursor_execute")
def receive_after_cursor_execute(
    conn, cursor, statement, parameters, context, executemany
):
    """Called after statement execution."""
    print(f"Execution complete")

Engine Utilities

Inspector

from sqlalchemy import inspect

inspector = inspect(engine)

# Get table names
tables = inspector.get_table_names()
print(f"Tables: {tables}")

# Get schema names
schemas = inspector.get_schema_names()
print(f"Schemas: {schemas}")

# Get columns
columns = inspector.get_columns('users')
for col in columns:
    print(f"{col['name']}: {col['type']}")

Dialect Information

# Get dialect
print(engine.dialect.name)  # postgresql, mysql, sqlite, etc.

# Check dialect features
print(engine.dialect.supports_native_boolean)  # True for PostgreSQL
print(engine.dialect.supports_sequences)  # True for PostgreSQL, Oracle
print(engine.dialect.max_identifier_length)  # 63 for PostgreSQL

# Server version
with engine.connect() as conn:
    version = conn.dialect.server_version_info
    print(f"Server version: {version}")

Common Patterns

Application Factory

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class Database:
    def __init__(self):
        self.engine = None
        self.Session = None
    
    def init(self, database_url, **kwargs):
        self.engine = create_engine(database_url, **kwargs)
        self.Session = sessionmaker(bind=self.engine)
    
    def get_session(self):
        return self.Session()

# Usage
db = Database()
db.init(
    'postgresql://user:pass@localhost/db',
    pool_size=10,
    pool_pre_ping=True
)

session = db.get_session()

Environment-Based Configuration

import os
from sqlalchemy import create_engine

def get_engine():
    database_url = os.getenv(
        'DATABASE_URL',
        'sqlite:///default.db'
    )
    
    # Development settings
    if os.getenv('ENV') == 'development':
        return create_engine(
            database_url,
            echo=True,
            pool_pre_ping=True
        )
    
    # Production settings
    return create_engine(
        database_url,
        pool_size=20,
        max_overflow=40,
        pool_recycle=3600,
        pool_pre_ping=True
    )

engine = get_engine()

Read Replicas

# Primary (write) engine
primary_engine = create_engine(
    'postgresql://user:pass@primary-host/db'
)

# Replica (read) engine
replica_engine = create_engine(
    'postgresql://user:pass@replica-host/db',
    execution_options={"read_only": True}
)

# Use primary for writes
with primary_engine.begin() as conn:
    conn.execute(users.insert().values(name='alice'))

# Use replica for reads
with replica_engine.connect() as conn:
    result = conn.execute(select(users))

SQL Expression Language

Schema Definition

Data Types

Engine & Connections

Engines

Engines

Creating Engines

Basic Engine Creation

URL Format

Database URLs

PostgreSQL

MySQL / MariaDB

SQLite

SQL Server

Oracle

URL Object

Programmatic URL Construction

Password Encoding

Parsing URLs

Engine Configuration

Common Parameters

Echo and Logging

Pool Configuration

Dialect-Specific Options

Connection Testing

Engine Disposal

Engine Events

Connection Events

Execution Events

Engine Utilities

Inspector

Dialect Information

Common Patterns

Application Factory

Environment-Based Configuration

Read Replicas

Next Steps

Connections

Connection Pooling

SQL Expression Language

Schema Definition

Data Types

Engine & Connections

​Engines

​Creating Engines

​Basic Engine Creation

​URL Format

​Database URLs

​PostgreSQL

​MySQL / MariaDB

​SQLite

​SQL Server

​Oracle

​URL Object

​Programmatic URL Construction

​Password Encoding

​Parsing URLs

​Engine Configuration

​Common Parameters

​Echo and Logging

​Pool Configuration

​Dialect-Specific Options

​Connection Testing

​Engine Disposal

​Engine Events

​Connection Events

​Execution Events

​Engine Utilities

​Inspector

​Dialect Information

​Common Patterns

​Application Factory

​Environment-Based Configuration

​Read Replicas

​Next Steps

Connections

Connection Pooling

Engines

Creating Engines

Basic Engine Creation

URL Format

Database URLs

PostgreSQL

MySQL / MariaDB

SQLite

SQL Server

Oracle

URL Object

Programmatic URL Construction

Password Encoding

Parsing URLs

Engine Configuration

Common Parameters

Echo and Logging

Pool Configuration

Dialect-Specific Options

Connection Testing

Engine Disposal

Engine Events

Connection Events

Execution Events

Engine Utilities

Inspector

Dialect Information

Common Patterns

Application Factory

Environment-Based Configuration

Read Replicas

Next Steps