Python的sqlalchemy库使用

neutron源码学习基础知识储备之Sqlalchemy库

本文主要参考官方文档和一些网上资料，并结合之前python-web-frame项目使用来详细说明Sqlalchemy的使用，版本号为SQLAlchemy 1.1。

Sqlalchemy的架构？

Object Relational Mapper && SQL Expression Language ?

下面是截取python-web-frame项目代码

#api.py
def get_engine():
    global _ENGINE
    if _ENGINE is not None:
        return _ENGINE
    _ENGINE = create_engine(
        "mysql+mysqldb://root:root123@10.0.38.237:3306/test?charset=utf8",
        echo=True)
    db_models.int_dbs(_ENGINE)
    return _ENGINE
# db_models.py
Base = declarative.declarative_base()
def int_dbs(_ENGINE):
    Base.metadata.create_all(_ENGINE)
class db_User(Base):
    __tablename__ = 'user'
    __table_args__ = (
        Index('ix_user_user_id', 'user_id'),
    )
    id = Column(Integer, primary_key=True)
    user_id = Column(String(255), nullable=False)
    name = Column(String(64), nullable=False, unique=True)
    gender = Column(String(64), nullable=False)
    age = Column(Integer, nullable=False)
    email = Column(String(255))
    def __repr__(self):
        return "<User(user_id='%s', name='%s', gender='%s',age='%s',email='%s')>" % (
            self.user_id, self.name, self.gender, self.age, self.email)
    def __init__(self, user_id, name, gender, age, email):
        self.user_id = user_id
        self.name = name
        self.gender = gender
        self.age = age
        self.email = email

Connecting

1
2
3

_ENGINE = create_engine(
        "mysql+mysqldb://root:root123@10.0.38.237:3306/test?charset=utf8",
        echo=True)

echo flag is a shortcut to setting up SQLAlchemy logging, which is accomplished via Python’s standard logging module. With it enabled, we’ll see all the generated SQL produced.

echo意思说开启日志，你可以看到整个SQL是如何产生的，方便调试。

_ENGINE is an instance of Engine, and it represents the core interface to the database, adapted through a dialect that handles the details of the database and DBAPI in use.

_ENGINE 意思说与数据库打交道的核心接口

Declare a Mapping

Base = declarative.declarative_base()
class db_User(Base):
    __tablename__ = 'user'
    __table_args__ = (
        Index('ix_user_user_id', 'user_id'),
    )
    id = Column(Integer, primary_key=True)
    user_id = Column(String(255), nullable=False)
    name = Column(String(64), nullable=False, unique=True)
    gender = Column(String(64), nullable=False)
    age = Column(Integer, nullable=False)
    email = Column(String(255))
    def __repr__(self):
        return "<User(user_id='%s', name='%s', gender='%s',age='%s',email='%s')>" % (
            self.user_id, self.name, self.gender, self.age, self.email)

生成一个映射使用的Base.

Create a Schema

1 2	def int_dbs(_ENGINE): Base.metadata.create_all(_ENGINE)

db_User类继承了Base类，它具有metadata属性，通过create_all()方法，注入与数据库打交道的核心接口_ENGINE，我们发现有一系列的命令完成数据库中user表是否存在的检测和生成。

Creating a Session

def get_session_maker(engine):
    global _SESSION_MAKER
    if _SESSION_MAKER is not None:
        return _SESSION_MAKER
    _SESSION_MAKER = sqlalchemy.orm.sessionmaker(bind=engine)
    return _SESSION_MAKER
def get_session():
    engine = get_engine()
    maker = get_session_maker(engine)
    session = maker()
    return session

This custom-made Session class will create new Session objects which are bound to our database.

Querying

http://docs.sqlalchemy.org/en/rel_1_1/orm/query.html#sqlalchemy.orm.query.Query

query()和 aliased()

Common Filter Operators

filter()

equals:
query.filter(User.name == 'ed')
not equals:
query.filter(User.name != 'ed')
LIKE:
query.filter(User.name.like('%ed%'))
IN:
query.filter(User.name.in_(['ed', 'wendy', 'jack']))
# works with query objects too:
query.filter(User.name.in_(
    session.query(User.name).filter(User.name.like('%ed%'))
))
NOT IN:
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
IS NULL:
query.filter(User.name == None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))
AND:
# use and_()
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
# or send multiple expressions to .filter()
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
# or chain multiple filter()/filter_by() calls
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones'
OR:
from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))
MATCH:
query.filter(User.name.match('wendy')

Returning Lists and Scalars

all() returns a list

first() applies a limit of one and returns the first result as a scalar

one() fully fetches all rows, and if not exactly one object identity or composite row is present in the result, raises an error

注意：The one() method is great for systems that expect to handle “no items found” versus “multiple items found” differently; such as a RESTful web service, which may want to raise a “404 not found” when no results are found, but raise an application error when multiple results are found.

one_or_none() is like one(), except that if no results are found, it doesn’t raise an error; it just returns None. Like one(), however, it does raise an error if multiple results are found

scalar() invokes the one() method, and upon success returns the first column of the row

Using Textual SQL

text()

Counting

count()

Building a Relationship

一对多

class db_User(Base):
    ....
    telephone = relationship("db_Telephone",order_by="db_Telephone.id",back_populates="user")
    ....
class db_Telephone(Base):
    ....
    user_id = Column(Integer, ForeignKey('user.id'))
    user = relationship("db_User", back_populates="telephone")

即：一个db_user对应多个db_Telephone

一对一

class db_User(Base):
    ....
    telephone = relationship("db_Telephone",uselist=False,back_populates="user")
    ....
class db_Telephone(Base):
    ....
    user_id = Column(Integer, ForeignKey('user.id'))
    user = relationship("db_User", back_populates="telephone")

多对多

Many to Many adds an association table between two classes.
多对多关系会在两个类之间增加一个关联的表。
The association table is indicated by the secondary argument to relationship().
这个关联的表在 relationship() 方法中通过 secondary 参数来表示。
Usually, the Table uses the MetaData object associated with the declarative base class,
通常的，这个表会通过 MetaData 对象来与声明基类关联，
so that the ForeignKey directives can locate the remote tables with which to link:
所以这个 ForeignKey 指令会使用链接来定位到远程的表：

#多对多关系中的两个表之间的一个关联表
post_keywords = Table('post_keywords', Base.metadata,
	Column('post_id', ForeignKey('posts.id'), primary_key=True),
	Column('keyword_id', ForeignKey('keywords.id'), primary_key=True)
                      
                      
class Parent(Base):
    __tablename__ = 'left'
    id = Column(Integer, primary_key=True)
    children = relationship(
        "Child",
        secondary=association_table,
        back_populates="parents")
class Child(Base):
    __tablename__ = 'right'
    id = Column(Integer, primary_key=True)
    parents = relationship(
        "Parent",
        secondary=association_table,
        back_populates="children")

Querying with Joins

Using Aliases

Using EXISTS

Common Relationship Operators

eq() (many-to-one “equals” comparison)

ne() (many-to-one “not equals” comparison)

IS NULL (many-to-one comparison, also uses eq())

contains() (used for one-to-many collections)

any() (used for collections)

has() (used for scalar references)

Query.with_parent() (used for any relationship)

Eager Loading

Query.options()

subqueryload()第一种

Joined Load()第二种

contains_eager()第三种

Deleting

db_user与db_Telephone是一对多关系，下面操作解决了删除db_user，会自动删除关联的表数据

def delete_user(self, user_id):
       logger.info("user.user_id: %s" % (user_id))
       try:
           session = get_session()
           user=session.query(
               db_models.db_User).filter_by(
               user_id=user_id).first()
           session.delete(user)
           session.flush()
           session.commit()
       except exc.NoResultFound:
           logger.error("delete user occur error ...")
class db_User(Base):
   ...
   telephone = relationship(
       "db_Telephone",
       order_by="db_Telephone.id",
       back_populates="user" ,
       cascade="save-update, merge, delete")
class db_Telephone(Base):
   ...
   user = relationship("db_User", back_populates="telephone")