SQLAlchemy ORM

Posted 2020-12-26 dontgiveup

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

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

1 mysql-Python    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
2   
3 pymysql    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
4   
5 MySQL-Connector    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
6   
7 cx_Oracle    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
8   
9 更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html

 

 1 import sqlalchemy
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
 4 from sqlalchemy.orm import sessionmaker, relationship
 5 from sqlalchemy import create_engine
 6 
 7 engine = create_engine("mysql+pymysql://root:[email protected]:3306/company",max_overflow=5);
 8 
 9 Base = declarative_base();
10 
11 class user(Base):
12     __tablename__ = "tableName":
13     id = Column(Integer,primary_key=True);
14     name = Column(String(16));
15     extra = Column(String(16));
16 
17     def __repr__(self):
18         pass
19 
20 #Base.metadata.create_all(engine); //创建所有关联的表
21 #Base.metadata.drop_all(engine); //删除所有关联的表

建表(create_all)

 

 1 import sqlalchemy
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
 4 from sqlalchemy.orm import sessionmaker, relationship
 5 from sqlalchemy import create_engine
 6 
 7 engine = create_engine("mysql+pymysql://root:[email protected]:3306/company",max_overflow=5);
 8 
 9 Base = declarative_base();
10 
11 class user(Base):
12     __tablename__ = "tableName":
13     id = Column(Integer,primary_key=True);
14     name = Column(String(16));
15     extra = Column(String(16));
16 
17     def __repr__(self):
18         pass
19 
20 #Base.metadata.create_all(engine); //创建所有关联的表
21 #Base.metadata.drop_all(engine); //删除所有关联的表
22 
23 Session = sessionmaker(bind=engine); //关联数据库
24 session = Session(); //创建session对象
25 
26 # 插入一条数据
27 session.add([user(name=‘name‘,extra=‘other‘),]);
28 
29 # 插入多条数据
30 session.add_all([user(name=‘name1‘,extra=‘other1‘),user(name=‘name2‘,extra=‘other2‘))]);
31 
32 session.commit(); #提交事务

增(add/add_all)

 

 1 engine = crement_engine(‘mysql+pymysql://root:[email protected]:3306/company‘);
 2 
 3 Session = sessionmaker(bind=engine);
 4 session = Session();
 5 
 6 result = session.query(User).filter(User.name.in_([‘xiaoming‘,‘xiaoyu‘])); # 查询一行数据
 7 
 8 result = session.query(User.name).filter(User.name.in_([‘xiaoming‘,‘xiaoyu‘])); #查询name列数据
 9 
10 #除了 in_ 还有 or_ , and_ 等等
11 
12 result = session.query(User).filter_by(User.name=‘xiaoyu‘);#查询指定数据
13 
14 正常的query操作返回的是一个mysql语句。而不是一个结果。需要有动作提交才会返回结果对象， 可能是一个列表，可能是一个类对象。 提交可以是 .all() .count() . 

查（query）

 

1 session.query(ClassName).filter(User.name=="xx")delete();
2 session.commit();

删( delete )

1 session.query(User).filter(User.id = 4).update({‘name‘:‘xxxxx});
2 session.commit();

改 (update)

 

 1 #　条件
 2 ret = session.query(Users).filter_by(name=‘alex‘).all()
 3 ret = session.query(Users).filter(Users.id > 1, Users.name == ‘eric‘).all()
 4 ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == ‘eric‘).all()
 5 ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
 6 ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
 7 ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=‘eric‘))).all()
 8 from sqlalchemy import and_, or_
 9 ret = session.query(Users).filter(and_(Users.id > 3, Users.name == ‘eric‘)).all()
10 ret = session.query(Users).filter(or_(Users.id < 2, Users.name == ‘eric‘)).all()
11 ret = session.query(Users).filter(
12     or_(
13         Users.id < 2,
14         and_(Users.name == ‘eric‘, Users.id > 3),
15         Users.extra != ""
16     )).all()
17 
18 
19 # 通配符
20 ret = session.query(Users).filter(Users.name.like(‘e%‘)).all()
21 ret = session.query(Users).filter(~Users.name.like(‘e%‘)).all()
22 
23 # 限制
24 ret = session.query(Users)[1:2]
25 
26 # 排序
27 ret = session.query(Users).order_by(Users.name.desc()).all()
28 ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
29 
30 # 分组
31 from sqlalchemy.sql import func
32 
33 ret = session.query(Users).group_by(Users.extra).all()
34 ret = session.query(
35     func.max(Users.id),
36     func.sum(Users.id),
37     func.min(Users.id)).group_by(Users.name).all()
38 
39 ret = session.query(
40     func.max(Users.id),
41     func.sum(Users.id),
42     func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
43 
44 # 连表
45 
46 ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
47 
48 ret = session.query(Person).join(Favor).all()
49 
50 ret = session.query(Person).join(Favor, isouter=True).all()
51 
52 
53 # 组合
54 q1 = session.query(Users.name).filter(Users.id > 2)
55 q2 = session.query(Favor.caption).filter(Favor.nid < 2)
56 ret = q1.union(q2).all()
57 
58 q1 = session.query(Users.name).filter(Users.id > 2)
59 q2 = session.query(Favor.caption).filter(Favor.nid < 2)
60 ret = q1.union_all(q2).all()

other

 

  1 #!/usr/bin/env python
  2 # -*- coding:utf-8 -*-
  3 from sqlalchemy import create_engine,and_,or_,func,Table
  4 from sqlalchemy.ext.declarative import declarative_base
  5 from sqlalchemy import Column, Integer, String,ForeignKey,UniqueConstraint,DateTime
  6 from  sqlalchemy.orm import sessionmaker,relationship
  7 
  8 Base = declarative_base() #生成一个SqlORM 基类
  9 
 10 # 服务器账号和组
 11 # HostUser2Group = Table(‘hostuser_2_group‘,Base.metadata,
 12 #     Column(‘hostuser_id‘,ForeignKey(‘host_user.id‘),primary_key=True),
 13 #     Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True),
 14 # )
 15 
 16 # 用户和组关系表，用户可以属于多个组，一个组可以有多个人
 17 UserProfile2Group = Table(‘userprofile_2_group‘,Base.metadata,
 18     Column(‘userprofile_id‘,ForeignKey(‘user_profile.id‘),primary_key=True),
 19     Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True),
 20 )
 21 
 22 # 程序登陆用户和服务器账户，一个人可以有多个服务器账号，一个服务器账号可以给多个人用
 23 UserProfile2HostUser= Table(‘userprofile_2_hostuser‘,Base.metadata,
 24     Column(‘userprofile_id‘,ForeignKey(‘user_profile.id‘),primary_key=True),
 25     Column(‘hostuser_id‘,ForeignKey(‘host_user.id‘),primary_key=True),
 26 )
 27 
 28 
 29 class Host(Base):
 30     __tablename__=‘host‘
 31     id = Column(Integer,primary_key=True,autoincrement=True)
 32     hostname = Column(String(64),unique=True,nullable=False)
 33     ip_addr = Column(String(128),unique=True,nullable=False)
 34     port = Column(Integer,default=22)
 35     def __repr__(self):
 36         return  "<id=%s,hostname=%s, ip_addr=%s>" %(self.id,
 37                                                     self.hostname,
 38                                                     self.ip_addr)
 39 
 40 
 41 class HostUser(Base):
 42     __tablename__ = ‘host_user‘
 43     id = Column(Integer,primary_key=True)
 44     AuthTypes = [
 45         (u‘ssh-passwd‘,u‘SSH/Password‘),
 46         (u‘ssh-key‘,u‘SSH/KEY‘),
 47     ]
 48     # auth_type = Column(ChoiceType(AuthTypes))
 49     auth_type = Column(String(64))
 50     username = Column(String(64),unique=True,nullable=False)
 51     password = Column(String(255))
 52 
 53     host_id = Column(Integer,ForeignKey(‘host.id‘))
 54     
 55     # groups = relationship(‘Group‘,
 56     #                       secondary=HostUser2Group,
 57     #                       backref=‘host_list‘)
 58 
 59     __table_args__ = (UniqueConstraint(‘host_id‘,‘username‘, name=‘_host_username_uc‘),)
 60 
 61     def __repr__(self):
 62         return  "<id=%s,name=%s>" %(self.id,self.username)
 63 
 64 
 65 class Group(Base):
 66     __tablename__ = ‘group‘
 67     id = Column(Integer,primary_key=True)
 68     name = Column(String(64),unique=True,nullable=False)
 69     def __repr__(self):
 70         return  "<id=%s,name=%s>" %(self.id,self.name)
 71 
 72 
 73 class UserProfile(Base):
 74     __tablename__ = ‘user_profile‘
 75     id = Column(Integer,primary_key=True)
 76     username = Column(String(64),unique=True,nullable=False)
 77     password = Column(String(255),nullable=False)
 78     # host_list = relationship(‘HostUser‘,
 79     #                       secondary=UserProfile2HostUser,
 80     #                       backref=‘userprofiles‘)
 81     # groups = relationship(‘Group‘,
 82     #                       secondary=UserProfile2Group,
 83     #                       backref=‘userprofiles‘)
 84     def __repr__(self):
 85         return  "<id=%s,name=%s>" %(self.id,self.username)
 86 
 87 
 88 class AuditLog(Base):
 89     __tablename__ = ‘audit_log‘
 90     id = Column(Integer,primary_key=True)
 91     userprofile_id = Column(Integer,ForeignKey(‘user_profile.id‘))
 92     hostuser_id = Column(Integer,ForeignKey(‘host_user.id‘))
 93     action_choices2 = [
 94         (u‘cmd‘,u‘CMD‘),
 95         (u‘login‘,u‘Login‘),
 96         (u‘logout‘,u‘Logout‘),
 97     ]
 98     action_type = Column(ChoiceType(action_choices2))
 99     #action_type = Column(String(64))
100     cmd = Column(String(255))
101     date = Column(DateTime)
102 
103     # user_profile = relationship("UserProfile")
104     #bind_host = relationship("BindHost")
105 
106 
107 engine = create_engine("mysql+pymsql://root:[email protected]:3306/stupid_jumpserver",echo=False)
108 Base.metadata.create_all(engine) #创建所有表结构
109 
110 表结构操作联系

表结构操作联系