介绍
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，
简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
1，安装：
pip3 install SQLAlchemy
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...] 更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
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
#连接mysql engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5) Base = declarative_base() # 创建单表 class Users(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(VARCHAR(32)) extra = Column(VARCHAR(16)) __table_args__ = ( UniqueConstraint('id', 'name', name='uix_id_name'), Index('ix_id_name', 'name', 'extra'), ) # 一对多 class Favor(Base): __tablename__ = 'favor' nid = Column(Integer, primary_key=True) caption = Column(VARCHAR(50), default='red', unique=True) class Person(Base): __tablename__ = 'person' nid = Column(Integer, primary_key=True) name = Column(VARCHAR(32), index=True, nullable=True) favor_id = Column(Integer, ForeignKey("favor.nid")) # 多对多 class Group(Base): __tablename__ = 'group' id = Column(Integer, primary_key=True) name = Column(VARCHAR(64), unique=True, nullable=False) port = Column(Integer, default=22) class Server(Base): __tablename__ = 'server' id = Column(Integer, primary_key=True, autoincrement=True) hostname = Column(VARCHAR(64), unique=True, nullable=False) 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')) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine)
2，操作表：
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:123@127.0.0.1:3306/t1", max_overflow=5) Base = declarative_base() # 创建单表 class Users(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(VARCHAR(32)) extra = Column(VARCHAR(16)) __table_args__ = ( UniqueConstraint('id', 'name', name='uix_id_name'), Index('ix_id_name', 'name', 'extra'), ) def __repr__(self): return "%s-%s" %(self.id, self.name) # 一对多 class Favor(Base): __tablename__ = 'favor' nid = Column(Integer, primary_key=True) caption = Column(VARCHAR(50), default='red', unique=True) def __repr__(self): return "%s-%s" %(self.nid, self.caption) 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")) # 与生成表结构无关，仅用于查询方便 favor = relationship("Favor", backref='pers') # 多对多 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')) group = relationship("Group", backref='s2g') server = relationship("Server", backref='s2g') class Group(Base): __tablename__ = 'group' id = Column(Integer, primary_key=True) name = Column(String(64), unique=True, nullable=False) port = Column(Integer, default=22) # group = relationship('Group',secondary=ServerToGroup,backref='host_list') class Server(Base): __tablename__ = 'server' id = Column(Integer, primary_key=True, autoincrement=True) hostname = Column(String(64), unique=True, nullable=False) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) Session = sessionmaker(bind=engine) session = Session()
3，案例：
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 ### 连接MySQL engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/db3?charset=utf8", max_overflow=5) Base = declarative_base() ''' create table users( id int auto_increment primary key, name varchar(32) not null default '', extra varchar(32) not null default '' ) ''' ''' Usertype id title xxoo 1 普通用户 row.xxoo : 多条记录对象 ''' class UserType(Base): __tablename__ = 'usertype' ### 表名 id = Column(Integer, autoincrement=True, primary_key=True) title = Column(String(32), nullable=False, server_default='') ''' users id name extra type_id 1 zekai nb 1 usertype = releationship('Usertype') row.usertype ''' class Users(Base): __tablename__ = 'users' id = Column(Integer, autoincrement=True, primary_key=True) name = Column(String(32), nullable=False, server_default='') extra = Column(String(32), nullable=False, server_default='') type_id = Column(Integer, ForeignKey(UserType.id)) usertype = relationship('UserType', backref='xxoo') #创建索引 __table_args__ = ( UniqueConstraint('id', 'name', name='uix_id_name'), ### 联合唯一索引 Index('ix_name_extra', 'name', 'extra'), ### 组合索引 ) # 删除表 def drop_db(): Base.metadata.drop_all(engine)
# 创建表 def create_db(): ## 会将当前执行文件中所有继承自Base类的类,生成表 Base.metadata.create_all(engine) drop_db() create_db() ###操作表中的数据 Session = sessionmaker(bind=engine) session = Session() ### 增加一条数据 obj = UserType(title='普通用户') session.add(obj) ### 添加多条数据 session.add_all([ UserType(title='VIP用户'), UserType(title='VIP中P用户'), UserType(title='SVIP用户'), UserType(title='黑金用户') ]) ### 查询 # ### 查询全部 返回的是一个列表, 列表中套对象 res = session.query(UserType).all() for row in res: print(row.id, row.name) #### 查询一条数据 res = session.query(UserType).first() print(res) # where条件 res = session.query(UserType).filter(UserType.name=='VIP用户', UserType.id==2).all() for row in res: print(row.id, row.name) print(res[0].name, res[0].id) res = session.query(UserType).filter_by(name='VIP用户').all() print(res) ## 删除: session.query(UserType).filter(UserType.id>3).delete()
## 修改 session.query(UserType).filter(UserType.id == 3).update({"name" : "SVIP用户"}) ### MySQL 高级查询操作 #### 通配符 分组 分页 排序 between and in not in ### between ... and ... res = session.query(UserType).filter(UserType.id.between(1,3)).all() for row in res: print(row.id, row.title)
## in 操作 bool_ res = session.query(UserType).filter(UserType.id.in_([1,3,4])).all() print(res)
### not in # select * from Usertype res = session.query(UserType).filter(~UserType.id.in_([1,3,4])).all() for row in res: print(row.id, row.title) session.query(UserType).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() res = session.query(UserType.title).filter(~UserType.id.in_([1,3,4])).all() for row in res: print( row.title) from sqlalchemy import and_, or_ ### 通配符 ret = session.query(UserType).all() from sqlalchemy.sql import func res = session.query( Users.type_id, func.max(Users.id), func.min(Users.id)).group_by(Users.type_id).all() print(res) ret = session.query( func.max(Users.id), func.min(Users.id)).group_by(Users.type_id).having(func.min(Users.id) >2).all() ### left join #############1. 查询某一个用户的用户类型##############
### 第一种方法: res = session.query(Users,UserType).join(UserType, isouter=True).all() # print(res) for row in res: print(row[0].id, row[0].name, row[1].title) ### 第二种方法 res = session.query(Users).all() for row in res: print(row.id, row.name, row.extra, row.usertype.title)
##############2. 某一个类型下面的用户###############
### 第一种方法 res = session.query(UserType).all() for row in res: print(row.id, row.title, session.query(Users).filter(Users.type_id == row.id).all() ) ### 第二种方法(正向|反向查询) res = session.query(UserType).all() for row in res: print(row.id, row.title, row.xxoo) session.commit() session.close()