数据库的那些ORM

说一下数据库CRUD的那些orm吧,其实从coding伊始,就开始各种orm。这里主要还是以python为例,从最开始的公司内部封装库,到mysqldb库2.x到3.x的变化,再到高级封装的sqlalchemmy,peewee等等。写这篇文章,主要还是现在工作又用到了sqlalchemy,我个人并不喜欢orm了,对sqlalchemy算是“深恶痛绝”?如果要用orm,我宁愿用peewee这些太轻巧型的。

orm其实就是对crud,session 以及sql注入等常见问题的封装处理。

反感orm,可能是经历过的项目代码对orm的封装处理实在是垃圾,而本人更喜欢写原生sql,sql调优真是一件快乐的事。喜欢orm的,没多少喜欢写原生sql语句的吧。

这里以py的sqlalchemy为例展开,sqlalchemy中主要的三种db方式:engine,connect,session。

engine和connect都是基于connect的封装。session不同的是内部加上了事务处理,官方推荐使用session。

db模型的创建通用两种:

一种是继承 db.Model 类。如下,可以在User类,添加各种函数方法,(个人不推荐,因为往往会越加越多,维护和可读都是成本啊)

#! /usr/bin/env python
# coding:utf8

import time
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


def current_ts():
    return int(time.time())


class User(db.Model):
    CONSTANT_SUPER = 1
    CONSTANT_ADMIN = 2
    CONSTANT_NORMAL = 3

    __table_args = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8"})

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), nullable=False, default="")
    status = db.Column(db.Integer, nullable=False, default=CONSTANT_NORMAL)
    create_time = db.Column(db.Integer, default=current_ts)
    update_time = db.Column(db.Integer, default=current_ts, onupdate=current_ts)

    @property
    def user_name(self):
        return ""

    @staticmethod
    def get_user_id(name):
        user = User.query.filter(User.name == name).first()
        return user.id

    """
    对于任何的学习成本和维护成本的增加,都是多余的
    """

另一种则是直接Table类实例,个人推荐,简单明了,相比上一种也能减少python类对象的创建消耗:

#! /usr/bin/env python
# coding:utf8
"""
personal recommend
pythonic if python
"""

from sqlalchemy import MetaData, Table, Column, Integer, text

meta = MetaData()
users_table = Table('users', meta,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)))

基于sqlalchemy的crud:

# 通用方法
db_records = User.query.filter
db_dict = {v.user_id: v.__dict__ for v in db_records}

分页处理:
pagination = records.paginate(args.page, args.limit, error_out=False)

 

# 个人推荐
# runs a transaction
with engine.begin() as connection:
    r1 = connection.execute(table1.select())
    connection.execute(table1.insert(), col1=7, col2='this is some data')

# runs a connection
with engine.connect() as conn:
    r = conn..execute(text("SELECT * FROM user WHERE id=:param"), {"param":5})
    r = conn..execute(text("SELECT * FROM user WHERE id=:param"), {"param":5}).fetchone()
    r = conn..execute(text("SELECT * FROM user WHERE id=:param"), {"param":5}).fetchall()

 

# 官方推荐seesion
session.execute(text("SELECT * FROM user WHERE id=:param"), {"param":5}).fetchone()

db_model的创建:

from sqlalchemy import (BigInteger, Boolean, Column, DateTime, Index, Integer, MetaData, Numeric, SmallInteger, String, Table, text, ARRAY)

t_base_info = Table(
    'base_info', metadata,
    Column('user_id', Integer, primary_key=True, server_default=text("nextval('base_info_user_id_seq'::regclass)")),
    Column('name', String(20), nullable=False, comment='用户昵称'),
    Column('email', String(128), nullable=False, comment='用户邮箱'),
    Column('area_code', String(6), nullable=False, comment='手机区号', server_default=text("'86'::character varying")),
    Column('phone', String(20), nullable=False, comment='用户手机'),
    Column('type', Integer, nullable=False, comment='用户类型,默认:普通用户', server_default=text("1")),
    Column('level', SmallInteger, nullable=False, comment='用户等级', server_default=text("'1'::smallint")),
    Column('is_valid', Boolean, nullable=False, server_default=text("true")),
    Column('create_time', DateTime, nullable=False, server_default=text("now()")),
)