Python 领域最著名的 ORM 框架，构建于第三方 DBAPI 之上

安装

pip install SQLAlchemy MySQL-python

使用

定义模型

from sqlalchemy import Column, SmallInteger, Integer, BigInteger, Float, CHAR, String, Date, DateTimefrom sqlalchemy.ext.declarative import declarative_baseBase = declarative_base() #模型基类class Test(Base):    __tablename__ = 'test'    id   = Column(Integer(unsigned=True), primary_key=True)    name = Column(Char(32))    desc = Column(String, nullable=True, index=True)

ORM调用

from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker# 数据库连接engine = create_engine('mysql+mysqldb://username:password@host:port/dbname')# 会话类Session = sessionmaker(bind=engine)# 会话实例session = Session()# 插入记录test_item = Test(id=1, name='aaa')session.add(test_item)session.commit()# 查询query = session.query(Test).filter(Test.id==1, ...).offset(10).limit(5).order_by('id asc')print query.statement #sql语句query.all|one|first() #对于first查询结果空返回Nonequery.get(id) #主键查询print test_item.namesession.commit()# 更新query.update({Test.desc: 'xxx'}) # 批量更新session.merge(new_test_item)) #主键搜索记录并更新session.commit()# 删除session.delete(test_item)session.commit()# 原生SQLsession.execute('select * from test where id = :id', {'id':1}).first()# 关闭会话session.close()

简单自定义封装

定义db.py模块

# -*- coding: utf-8 -*-""" ORM公共层 """import osimport typesfrom sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmakerfrom sqlalchemy.ext.declarative import declarative_base""" 数据库连接 """_con_params = (    'mysql',    'mysqldb',    os.environ.get('MYSQL_USERNAME'),    os.environ.get('MYSQL_PASSWORD'),    os.environ.get('MYSQL_HOST'),    os.environ.get('MYSQL_PORT'),    os.environ.get('MYSQL_DATABASE'),)_engine = create_engine('%s+%s://%s:%s@%s:%s/%s' % _con_params)_session_factory = sessionmaker(bind=_engine)session = _session_factory()""" 模型基类 """class ModelMixin:    @classmethod    def query(cls):        return session.query(cls)    @classmethod    def create(cls, data):        if type(data) ==  types.DictType:            record = cls(**data)        else:            record = data        session.add(record)        session.commit()        return reload    @classmethod    def find(cls, id):        return cls.query().get(id)    def save(self):        session.merge(self)        session.commit()BaseModel = declarative_base()

参考

• 列定义：
• 列类型：
• Session API：
• Query API：