2019独角兽企业重金招聘Python工程师标准>>>
Python 领域最著名的 ORM 框架,构建于第三方 DBAPI 之上
安装
pip install SQLAlchemy MySQL-python
使用
定义模型
from sqlalchemy import Column, SmallInteger, Integer, BigInteger, Float, CHAR, String, Date, DateTime
from sqlalchemy.ext.declarative import declarative_base
Base = 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_engine
from 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查询结果空返回None
query.get(id) #主键查询
print test_item.name
session.commit()
# 更新
query.update({Test.desc: 'xxx'}) # 批量更新
session.merge(new_test_item)) #主键搜索记录并更新
session.commit()
# 删除
session.delete(test_item)
session.commit()
# 原生SQL
session.execute('select * from test where id = :id', {'id':1}).first()
# 关闭会话
session.close()
简单自定义封装
定义db.py模块
# -*- coding: utf-8 -*-
""" ORM公共层 """
import os
import types
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from 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()
参考
- 列定义:https://docs.sqlalchemy.org/en/rel_1_1/core/metadata.html#sqlalchemy.schema.Column
- 列类型:https://docs.sqlalchemy.org/en/rel_1_1/core/type_basics.html
- Session API:http://docs.sqlalchemy.org/en/rel_1_1/orm/session_api.html#sqlalchemy.orm.session.Session.add
- Query API:http://docs.sqlalchemy.org/en/rel_1_1/orm/query.html#sqlalchemy.orm.query.Query