Using SQLAlchemy ORM in Python: Design, Initialization, Model Mapping, and Session Management

The author originally used pymysql but switched to an ORM framework to avoid maintenance headaches and injection risks, choosing SQLAlchemy for its powerful object‑relational mapping capabilities.

Initialization : Create a MySQL engine with create_engine('mysql+pymysql://$user:$password@$host:$port/$db?charset=utf8mb4', echo=True) and set up a base declarative model.

<code>from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://$user:$password@$host:$port/$db?charset=utf8mb4', echo=True)</code>

Project Design : Follow an MVC‑like layout where model holds the ORM classes and model_op provides CRUD operations, keeping the business logic decoupled from the data layer.

<code>├── main.py
├── model
│   ├── __init__.py
│   ├── base_model.py
│   ├── ddl.sql
│   └── py_orm_model.py
└── model_op
    ├── __init__.py
    └── py_orm_model_op.py</code>

Model Declaration : Define a Python class that maps to a MySQL table, specifying columns, primary keys, defaults, and comments.

<code># py_orm_model.py
from .base_model import Base
from sqlalchemy import Column, Integer, String, TIMESTAMP, text, JSON

class PyOrmModel(Base):
    __tablename__ = 'py_orm'
    id = Column(Integer, autoincrement=True, primary_key=True, comment='唯一id')
    name = Column(String(255), nullable=False, default='', comment='名称')
    attr = Column(JSON, nullable=False, comment='属性')
    ct = Column(TIMESTAMP, nullable=False, server_default=text('CURRENT_TIMESTAMP'), comment='创建时间')
    ut = Column(TIMESTAMP, nullable=False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'), comment='更新时间')
</code>

The Base class is created with declarative_base() and the engine is shared across modules.

<code># base_model.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
engine = create_engine("mysql+pymysql://root:[email protected]:33306/orm_test?charset=utf8mb4", echo=False)
</code>

Session Management : Use a context manager to obtain, commit, rollback, and close sessions automatically.

<code># base_model.py (session utilities)
from contextlib import contextmanager
from sqlalchemy.orm import sessionmaker, scoped_session

def _get_session():
    """Retrieve a scoped session"""
    return scoped_session(sessionmaker(bind=engine, expire_on_commit=False))()

@contextmanager
def db_session(commit=True):
    session = _get_session()
    try:
        yield session
        if commit:
            session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        if session:
            session.close()
</code>

Utility methods are added to the model for converting between objects and dictionaries.

<code># py_orm_model.py (conversion helpers)
@staticmethod
def fields():
    return ['id', 'name', 'attr']

@staticmethod
def to_json(model):
    json_data = {}
    for field in PyOrmModel.fields():
        json_data[field] = getattr(model, field)
    return json_data

@staticmethod
def from_json(data: dict):
    model = PyOrmModel()
    for field in PyOrmModel.fields():
        if field in data:
            setattr(model, field, data[field])
    return model
</code>

Database Operations : The PyOrmModelOp class wraps CRUD actions using the session manager, so callers only need to pass plain dictionaries.

<code># py_orm_model_op.py
from sqlachlemy_lab.model import db_session
from sqlachlemy_lab.model import PyOrmModel

class PyOrmModelOp:
    @staticmethod
    def save_data(data: dict):
        with db_session() as session:
            model = PyOrmModel.from_json(data)
            session.add(model)

    @staticmethod
    def query_data(pid: int):
        data_list = []
        with db_session(commit=False) as session:
            rows = session.query(PyOrmModel).filter(PyOrmModel.id == pid)
            for d in rows:
                data_list.append(PyOrmModel.to_json(d))
        return data_list
</code>

Usage Example demonstrates inserting a record via the operation class.

<code># main.py
from sqlachlemy_lab.model_op import PyOrmModelOp

if __name__ == '__main__':
    PyOrmModelOp.save_data({'id': 1, 'name': 'test', 'attr': {}})
</code>

Running the script triggers SQLAlchemy to emit the corresponding CREATE TABLE DDL and perform the insert, as shown in the log output.