python下sqlalchemy的orm初探
项目发展到一定阶段,总是难免要与数据库打交道。手写sql语句比较笨拙,也容易出错,而且换数据库的移植性还不太好。
咋整,用成熟的orm轮子吧。
ORM(Object-relational mapping)
权威的概念定义可以参考维基百科。
具体啥作用呢,简单的说就是用类的实例的方法调用来实现CRUD。调用相应方法就可以在数据库中增加一条记录,要修改某条记录只需要修改类实例对象的参数就行。
Sqlalchemy
之前用django框架里的orm,感觉很好用,直接create,然后用实例调用save就可以存到数据库中。
原始的sqlalchemy要创建表、创建修改记录还需要有session的参与,没有直接的create、save函数,需要自己封装或者添加一些“插件”。
sqlalchemy的插件这里推荐两个:sqlalchemy_utils,sqlalchemy_mixins。
这两个插件其实就是对sqlalchemy的功能补充。utils添加了一些方便的函数,比如判断数据库是否存在,一条语句创建数据库。mixins就是添加了create、save的方法接口,可以更方便面向对象式使用。
1.sqlalchemy_utils
数据库不存在则创建相应数据库。
from sqlalchemy_utils.functions import database_exists, create_database
DB_NAME = "stock_data"
db_url = {
'database': DB_NAME,
'drivername': 'mysql',
'username': 'root',
'password': '111111',
'host': '127.0.0.1',
"query": {"charset": "utf8"},
}
engine = create_engine(URL(**db_url), encoding="utf8")
# create database if not exists
if not database_exists(engine.url):
create_database(engine.url)
2.sqlalchemy_mixins
实现面向对像类似django-orm的方式来进行crud。
from sqlalchemy_mixins import AllFeaturesMixin
class TestMixin(Base, AllFeaturesMixin):
__tablename__ = "test_mixins"
id = Column(Integer, primary_key=True)
name = Column(VARCHAR(200))
date = Column(DateTime, default=datetime.utcnow)
TestMixin.create(id=2,name="second")
3. 手写CRUD基类方法
class CRUD:
def save(self):
if self.id is not None:
session.add(self)
return session.commit()
def destroy(self):
session.delete(self)
return session.commit()
这个类并不完善,只是一个demo给出大概思路:)。在基类实现save、destroy的方法,子类去继承重用就可以。
扫描二维码关注公众号,回复:
2721008 查看本文章
4. sqlalchemy原生版的保存记录代码
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
data = {"a": 5566, "b": 334, "c": 3232}
try:
for _key, _val in data.items():
row = TestBase(key=_key, val=_val)
session.add(row)
session.commit()
except SQLAlchemyError as e:
print(e)
session.close()
完整代码
from datetime import datetime
from sqlalchemy import create_engine, Column, Integer, String, DateTime, VARCHAR
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine.url import URL
from sqlalchemy_utils.functions import database_exists, create_database
from sqlalchemy_mixins import AllFeaturesMixin
DB_NAME = "stock_data"
db_url = {
'database': DB_NAME,
'drivername': 'mysql',
'username': 'root',
'password': '111111',
'host': '127.0.0.1',
"query": {"charset": "utf8"}, # 中文环境下非常关键的设置
}
engine = create_engine(URL(**db_url), encoding="utf8")
# create database if not exists
if not database_exists(engine.url):
create_database(engine.url)
Base = declarative_base()
class TestBase(Base):
__tablename__ = 'test table'
id = Column(Integer, primary_key=True)
key = Column(VARCHAR(200), nullable=False)
val = Column(VARCHAR(200))
date = Column(DateTime, default=datetime.utcnow)
class TestMixin(Base, AllFeaturesMixin):
__tablename__ = "test_mixins"
id = Column(Integer, primary_key=True)
name = Column(VARCHAR(200))
date = Column(DateTime, default=datetime.utcnow)
# create tables
Base.metadata.create_all(bind=engine)
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
# 此处调用后,所有派生类都不需要设置session了。否则每个派生类都要调用一次set_session函数
AllFeaturesMixin.set_session(session)
class CRUD:
def save(self):
if self.id is not None:
session.add(self)
return session.commit()
def destroy(self):
session.delete(self)
return session.commit()
def save_record(*params):
for k in params:
session.add(k)
session.commit()
def create_all_tables():
Base.metadata.create_all(bind=engine)
def drop_all_tables():
Base.metadata.drop_all(bind=engine)
def commit_all_flush():
try:
session.commit()
except:
session.rollback()
if __name__ == '__main__':
TestMixin.create(id=3, name="some")
commit_all_flush()
data = {"a": 5566, "b": 334, "c": 3232}
try:
for _key, _val in data.items():
row = TestBase(key=_key, val=_val)
session.add(row)
session.commit()
except SQLAlchemyError as e:
print(e)
session.close()
PS
- mysql命令行中带空格的表名用反引号包含起来,否则命令行中会提示找不到对应的表:)。
- 设置数据库连接的字符集。中文环境下非常关键。:)。不设定,列中输入中文时会报错的:)。
db_url = {
'database': DB_NAME,
'drivername': 'mysql',
'username': 'root',
'password': '111111',
'host': '127.0.0.1',
"query": {"charset": "utf8"}, # 中文环境下非常关键的设置
}
如果没有设定字符集,会报类似下面的错误
TestMixin.create(id=4, name="中文字符引发错误") # 此处的中文字符会引发错误
Error: 'latin1' codec can't decode byte 0xc3 in position 5