ORM 框架

逃离我推掉我的手 2022-02-02 04:57 346阅读 0赞

#### ORM 框架 ####

转：https://www.cnblogs.com/pycode/p/mysql-orm.html

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzI3ODA4OQ_size_16_color_FFFFFF_t_70]

SQLAlchemy本身无法操作数据库，其必须依赖pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

MySQL-Python
        mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
      
    pymysql
        mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
      
    MySQL-Connector
        mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
      
    cx_Oracle
        oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

底层处理

使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。

from sqlalchemy import create_engine
    
    #创建引擎
    engine = create_engine("mysql+pymysql://lx:123456@127.0.0.1:3306/Role", max_overflow=5)
    #执行sql语句
    engine.execute("INSERT INTO user (name) VALUES ('lx')")
    
    result = engine.execute('select * from user')
    res = result.fetchall()
    print(res)

ORM功能使用  
使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。

#### 创建表 ####

from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
    from sqlalchemy.orm import sessionmaker, relationship
    from sqlalchemy import create_engine
    
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/lx?charset=utf8", max_overflow=5,encoding='utf-8') 
    Base = declarative_base()
    
    # 创建单表
    class Users(Base):
        __tablename__ = 'users'
        id = Column(Integer, primary_key=True)
        name = Column(String(32))
    	extra = Column(String(16))
    
    # 一对多
    class Favor(Base):
        __tablename__ = 'favor'
        nid = Column(Integer, primary_key=True)
        caption = Column(String(50), default='red', unique=True)
    
    class Person(Base):
        __tablename__ = 'person'
        nid = Column(Integer, primary_key=True)
        name = Column(String(32), index=True, nullable=True)
        favor_id = Column(Integer, ForeignKey("favor.nid"))
    
    # 多对多
    class ServerToGroup(Base):
        __tablename__ = 'servertogroup'
        nid = Column(Integer, primary_key=True, autoincrement=True)
        server_id = Column(Integer, ForeignKey('server.id'))
        group_id = Column(Integer, ForeignKey('group.id'))
    
    class Group(Base):
        __tablename__ = 'group'
        id = Column(Integer, primary_key=True)
        name = Column(String(64), unique=True, nullable=False)
    
    class Server(Base):
        __tablename__ = 'server'
        id = Column(Integer, primary_key=True, autoincrement=True)
        hostname = Column(String(64), unique=True, nullable=False)
        port = Column(Integer, default=22)
    
    Base.metadata.create_all(engine)  #创建表
    # Base.metadata.drop_all(engine) #删除表
    
    Session = sessionmaker(bind=engine)
    session = Session()

*  增

obj = Users(name="alex0", extra='a1')
    session.add(obj)
    session.add_all([
        Users(name="alex1", extra='a2'),
        Users(name="alex2", extra='a3')
    ])
    session.commit()

*  删

session.query(Users).filter(Users.id > 2).delete()
    session.commit()

*  改

session.query(Users).filter(Users.id > 2).update({ "name" : "099"})
    session.query(Users).filter(Users.id > 2).update({ Users.name: Users.name + "099"}, synchronize_session=False)
    session.query(Users).filter(Users.id > 2).update({ "num": Users.num + 1}, synchronize_session="evaluate")
    session.commit()

*  查

ret = session.query(Users).all()
    ret = session.query(Users.name, Users.extra).all()
    ret = session.query(Users).filter_by(name='alex').all()
    ret = session.query(Users).filter_by(name='alex').first()

*  其他

#　条件
    ret = session.query(Users).filter_by(name='alex').all()
    ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
    ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
    ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
    ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
    # 嵌套查询
    ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
    
    # and_, or_
    from sqlalchemy import and_, or_
    ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
    ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
    ret = session.query(Users).filter(
        or_(
            Users.id < 2,
            and_(Users.name=='eric', Users.id > 3),
            Users.extra != ""
        )).all()
    
    
    # 通配符
    ret = session.query(Users).filter(Users.name.like('e%')).all()
    ret = session.query(Users).filter(~Users.name.like('e%')).all()
    
    # 限制
    ret = session.query(Users)[1:2]
    
    # 排序
    ret = session.query(Users).order_by(Users.name.desc()).all()
    ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
    
    # 分组
    from sqlalchemy.sql import func
    
    ret = session.query(Users.extra, func.count('*')).group_by(Users.extra).all()
    ret = session.query(
        func.max(Users.id),
        func.sum(Users.id),
        func.min(Users.id)).group_by(Users.name).all()
    
    ret = session.query(
        func.max(Users.id),
        func.sum(Users.id),
        func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
    
    # 连表
    
    ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
    
    # 内连接
    ret = session.query(Person).join(Favor).all()
    
    # 外连接
    ret = session.query(Person).join(Favor, isouter=True).all()
    
    
    # 组合 union联合查询
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union(q2).all()
    
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union_all(q2).all()

*  ORM解决中文编码问题 sqlalchemy 默认使用latin-1进行编码。所以当出现中文时就会报如下错误：

UnicodeEncodeError: 'latin-1' codec can't encode characters in position 39-41: ordinal not in range(256)

解决方法：

在连接数据库的时候直接指定字符编码：

engine = create_engine("mysql+pymysql://lx:123456@127.0.0.1:3306/Role?charset=utf8", max_overflow=5,encoding='utf-8')

*  ORM 指定查询返回数据格式 默认使用query查询返回的结果为一个对象

res = session.query(User).all()
    print(res)
    
    输出结果：
    [<__main__.User object at 0x10385c438>, <__main__.User object at 0x10385c4a8>, <__main__.User object at 0x10385c550>, <__main__.User object at 0x10385c5f8>, <__main__.User object at 0x10385c6a0>]

解决方法：

使用\_\_repr\_\_定义返回的数据

class User(Base):
        __tablename__ = 'user'
        nid = Column(Integer,primary_key=True,autoincrement=True)
        name = Column(String(10),nullable=False)
        
        def __repr__(self):
            output = "(%s,%s,%s)" %(self.nid,self.name,self.role)
            return output
            
    res = session.query(User).all()
    print(res)
    
    输出：
    [(1,fuzj,1), (2,jie,2), (3,张三,2), (4,李四,1), (5,王五,3)]

##### 一对多 #####

实例

*  ORM一对多具体使用
 *  设计两张表，user表和role表
 *  user 表中存放用户，role表中存放用户角色，role表中角色对应user表中多个用户，user表中一个用户只对应role表中一个角色，中间通过外键约束

from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String,ForeignKey
    from sqlalchemy.orm import sessionmaker,relationship
    from sqlalchemy import create_engine
    
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/lx02?charset=utf8", max_overflow=5,encoding='utf-8') 
    Base = declarative_base()
    
    class Role(Base):
        __tablename__ = 'role'
        rid = Column(Integer, primary_key=True, autoincrement=True)    #主键，自增
        role_name = Column(String(10))
    
        def __repr__(self):
            output = "(%s,%s)" %(self.rid,self.role_name)
            return output
    
    class User(Base):
        __tablename__ = 'user'
        nid = Column(Integer,primary_key=True,autoincrement=True)
        name = Column(String(10),nullable=False)
        role_id = Column(Integer,ForeignKey('role.rid'))  #外键关联
    
        def __repr__(self):
            output = "(%s,%s,%s)" %(self.nid,self.name,self.role_id)
            return output
    Base.metadata.create_all(engine)
    
    Session = sessionmaker(bind=engine)
    session = Session()
    
    #添加角色数据
    session.add(Role(role_name='dba'))
    session.add(Role(role_name='sa'))
    session.add(Role(role_name='net'))
    
    #添加用户数据
    session.add_all([
        User(name='小红',role='1'),
        User(name='小明',role='2'),
        User(name='张三',role='2'),
        User(name='李四',role='1'),
        User(name='王五',role='3'),
    ])
    session.commit()
    session.close()

*  普通连表查询

res = session.query(User,Role).join(Role).all()    #查询所有用户,及对应的role id
    res1 = session.query(User.name,Role.role_name).join(Role).all()  #查询所有用户和角色,
    res2 = session.query(User.name,Role.role_name).join(Role,isouter=True).filter(Role.role_name=='sa').all() #查询所有sa的用户
    print(res)
    print(res1)
    print(res2)
    
    输出结果：
    [((1,小红,1), (1,dba)), ((2,小明,2), (2,sa)), ((3,张三,2), (2,sa)), ((4,李四,1), (1,dba)), ((5,王五,3), (3,net))]
    [('小红', 'dba'), ('小明', 'sa'), ('张三', 'sa'), ('李四', 'dba'), ('王五', 'net')]
    [('jie', 'sa'), ('张三', 'sa')]

##### 使用relationship 添加影射关系进行查询（反向查询） #####

*  首先在User表中添加relationship映射关系

class Role(Base):
        __tablename__ = 'role'
        rid = Column(Integer, primary_key=True, autoincrement=True)    #主键，自增
        role_name = Column(String(10))
    
        def __repr__(self):
            output = "(%s,%s)" %(self.rid,self.role_name)
            return output
    
    class User(Base):
        __tablename__ = 'user'
        nid = Column(Integer,primary_key=True,autoincrement=True)
        name = Column(String(10),nullable=False)
        role_id = Column(Integer,ForeignKey('role.rid'))
        role = relationship("Role",backref='user')    #Role为类名
        
        def __repr__(self):
            output = "(%s,%s,%s)" %(self.nid,self.name,self.role)
            return output

*  查询

#正向查询
    print('正向查询')
    res = session.query(User).all()  #查询所有的用户和角色
    for u in res:
        print(u.name,u.role.role_name)    #此时的u.role 就是role表对应的关系
    res = session.query(User).filter(User.name=='小明').first()  #查询fuzj用户和角色
    print(res.name,res.role.role_name)
    
    print('反向查找')
    #反向查找
    res = session.query(Role).filter(Role.role_name =='dba').first()   #查找dba组下的所有用户
    print(res.user)   #此时 print的结果为[(1,小明,1), (4,李四,1)]
    for i in res.user:
        print(i.name,res.role_name)
    
    输出结果：
    正向查询
    小明 dba
    小红 sa
    张三 sa
    李四 dba
    王五 net
    小明 dba
    反向查找
    [(1,小明,1), (4,李四,1)]
    小明 dba
    李四 dba

*  说明

relationship 在user表中创建了新的字段，这个字段只用来存放user表中和role表中的对应关系，在数据库中并不实际存在  
所谓正向和反向查找是对于relationship关系映射所在的表而说，如果通过该表（user表）去查找对应的关系表（role表），就是正向查找，反正通过对应的关系表（role表）去查找该表（user表）即为反向查找。而relationship往往会和ForeignKey共存在一个表中。

##### ORM多对多 #####

实例

Mysql多对多关系指的是两张表A和B本没有任何关系，而是通过第三张表C建立关系，通过关系表C，使得表A在表B中存在多个关联数据，表B在表A中同样存在多个关联数据

*  创建三张表 host表 hostuser表 host\_to\_hostuser表
 *  host表中存放主机，hostuser表中存放主机的用户， host\_to\_hostuser表中存放主机用户对应的主机，hostuser表中用户对应host表中多个主机，host表中主机对应hostuser表中多个用户，中间关系通过host\_to\_hostuser表进行关联。host\_to\_hostuser和host表、user表进行外键约束

from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String,ForeignKey
    from sqlalchemy.orm import sessionmaker,relationship
    from sqlalchemy import create_engine
    
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/lx03?charset=utf8", max_overflow=5,encoding='utf-8') 
    Base = declarative_base()
    
    class Host(Base):
        __tablename__ = 'host'
        nid = Column(Integer, primary_key=True,autoincrement=True)
        hostname = Column(String(32))
        port = Column(String(32))
        ip = Column(String(32))
    
    class HostUser(Base):
        __tablename__ = 'host_user'
        nid = Column(Integer, primary_key=True,autoincrement=True)
        username = Column(String(32))
    
    class HostToHostUser(Base):
        __tablename__ = 'host_to_host_user'
        nid = Column(Integer, primary_key=True,autoincrement=True)
        host_id = Column(Integer,ForeignKey('host.nid'))
        host_user_id = Column(Integer,ForeignKey('host_user.nid'))
    
    Base.metadata.create_all(engine)
    
    Session = sessionmaker(bind=engine)
    session = Session()
    
    #添加数据
    session.add_all([
        Host(hostname='c1',port='22',ip='1.1.1.1'),
        Host(hostname='c2',port='22',ip='1.1.1.2'),
        Host(hostname='c3',port='22',ip='1.1.1.3'),
        Host(hostname='c4',port='22',ip='1.1.1.4'),
        Host(hostname='c5',port='22',ip='1.1.1.5'),
    ])
    
    session.add_all([
        HostUser(username='root'),
        HostUser(username='db'),
        HostUser(username='nb'),
        HostUser(username='sb'),
    ])
    
    session.add_all([
        HostToHostUser(host_id=1,host_user_id=1),
        HostToHostUser(host_id=1,host_user_id=2),
        HostToHostUser(host_id=1,host_user_id=3),
        HostToHostUser(host_id=2,host_user_id=2),
        HostToHostUser(host_id=2,host_user_id=4),
        HostToHostUser(host_id=2,host_user_id=3),
    ])
    
    session.commit()
    session.close()

*  普通多次查询

第一步：host_id = session.query(Host.nid).filter(Host.hostname=='c2').first()   #查找hostbane对应的hostid,返回结果为元组(2,)
    
    第二步： user_id_list = session.query(HostToHostUser.host_user_id).filter(HostToHostUser.host_id==host_id[0]).all()  #查询hostid对应的所有userid
    
    user_id_list = zip(*user_id_list)   #user_id_list 初始值为[(2,), (4,), (3,)],使用zip转换为[2,4,3]对象
    #print(list(user_id_list)) #结果为[(2, 4, 3)]
    
    第三步：user_list = session.query(HostUser.username).filter(HostUser.nid.in_(list(user_id_list)[0])).all()  #查询符合条件的用户
    print(user_list)
    
    #或者：
    user_id_list = session.query(HostToHostUser.host_user_id).join(Host).filter(Host.hostname=='c2').all()
    user_id_list = zip(*user_id_list)
    user_list = session.query(HostUser.username).filter(HostUser.nid.in_(list(user_id_list)[0])).all()
    print(user_list)
    
    # 输出结果
    [('db',), ('nb',), ('sb',)]

##### 使用relationship映射关系查询 #####

*  首先在关系表Host\_to\_hostuser中加入relationship关系映射

class HostToHostUser(Base):
        __tablename__ = 'host_to_host_user'
        nid = Column(Integer, primary_key=True,autoincrement=True)
    
        host_id = Column(Integer,ForeignKey('host.nid'))
        host_user_id = Column(Integer,ForeignKey('host_user.nid'))
        host = relationship('Host',backref='h') #对应host表
        host_user = relationship('HostUser',backref='u') #对应host_user表

*  查询

#查找一个服务器上有哪些用户
    res = session.query(Host).filter(Host.hostname=='c2').first()  #返回的是符合条件的服务器对象
    res2 = res.h    #通过relationship反向查找 Host_to_Hostuser中的对应关系
    for i in res2:   #i为host_to_hostuser表和host表中c2主机有对应关系的条目
        print(i.host_user.username)        #正向查找, 通过relationship ,找到host_to_hostuser中对应的hostuser 即i.host_user
    
    #查找此用户有哪些服务器
    res = session.query(HostUser).filter(HostUser.username=='sb').first()
    for i in res.u:
        print(i.host.hostname)

##### 扩展查询 #####

*  不查询关系表（中间表），直接在hostuser表中指定关系表，然后获取host表
 *  在host表中使用 relationship的secondary指定关系表。

class Host(Base):
        __tablename__ = 'host'
        nid = Column(Integer, primary_key=True,autoincrement=True)
        hostname = Column(String(32))
        port = Column(String(32))
        ip = Column(String(32))
        host_user = relationship('HostUser',secondary=lambda :HostToHostUser.__table__,backref='h')
        
    
    查询：
    
    host_obj = session.query(Host).filter(Host.hostname=='c1').first()
    for i in host_obj.host_user:
        print(i.username)

#### 创建一个实时监控文件系统 ####

我将使用 SQLAlchemy 演示如何创建一个元数据工具。此工具的目标是监控文件系统、创建和删除事件，以及在一个 SQLAlchemy 数据库中保存这些变更的记录

#/usr/bin/env python2.7
    
    import os
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.orm import scoped_session
     
    from pyinotify import *
     
    path = "/tmp"
     
    #SQLAlchemy
    engine = create_engine('sqlite:///meta.db', echo=True)
    Base = declarative_base()
    Session = scoped_session(sessionmaker(bind=engine))
     
    class Filesystem(Base):
        __tablename__ = 'filesystem'
        path = Column(String, primary_key=True)
        name = Column(String)
     
        def __init__(self, path,name):
            self.path = path
            self.name = name
     
        def __repr__(self):
            return "<Metadata('%s','%s')>" % (self.path,self.name)
     
    # 装饰器
    def transactional(fn):
        """add transactional semantics to a method."""
     
        def transact(self, *args):
            session = Session()
            try:
                fn(self, session, *args)
                session.commit()
            except:
                session.rollback()
                raise
        transact.__name__ = fn.__name__
        return transact
     
     
    class ProcessDir(ProcessEvent):
        """Performs Actions based on mask values"""
     
        @transactional
        def process_IN_CREATE(self, session, event):
            print "Creating File and File Record:", event.pathname
            create_record = Filesystem(event.pathname, event.path)
            session.add(create_record)
     
        @transactional
        def process_IN_DELETE(self, session, event):
            print "Removing:", event.pathname
            delete_record = session.query(Filesystem).\
                filter_by(path=event.pathname).one()
            session.delete(delete_record)
     
    def init_repository():
        #Drop the table, then create again with each run
        Base.metadata.drop_all(engine)
        Base.metadata.create_all(engine)
        session = Session()
     
        #Initial Directory Walking Addition Brute Force
        for dirpath, dirnames, filenames in os.walk(path):
            for file in filenames:
                fullpath = os.path.join(dirpath, file)
                record = Filesystem(fullpath, file)
                session.add(record)
            session.flush()
     
        for record in session.query(Filesystem):
            print "Database Record Number: Path: %s , File: %s " \
            % (record.path, record.name)
     
        session.commit()
     
    if __name__ ==  "__main__":
     
        init_repository()
     
        #Pyionotify
        wm = WatchManager()
        mask = IN_DELETE | IN_CREATE
        notifier = ThreadedNotifier(wm, ProcessDir())
        notifier.start()
     
        wdd = wm.add_watch(path, mask, rec=True)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzI3ODA4OQ_size_16_color_FFFFFF_t_70]: /images/20220202/a71899cb6ce74d4ca824260a0c5c4fe3.png