SQLAlchemy安装
- 安装
- pip install sqlalchemy -i https://pypi.douban.com/simple
- 连接的时候依赖pymysql
通过SQLAlchemy连接数据库
- from sqlalchemy import create_engine
- # 数据库的配置变量
- HOSTNAME = '127.0.0.1'
- PORT = '3306'
- DATABASE = 'xt_flask'
- USERNAME = 'root'
- PASSWORD = 'root'
- DB_URI = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(USERNAME,PASSWORD,HOSTNAME,PORT,DATABASE)
- # 创建数据库引擎
- engine = create_engine(DB_URI)
- #创建连接
- with engine.connect() as con:
- rs = con.execute('SELECT 1')
- print rs.fetchone()
- 首先从sqlalchemy中导入create_engine，用这个函数来创建引擎，然后用engine.connect()来连接数据库。其中一个比较重要的一点是，通过create_engine函数的时候，需要传递一个满足某种格式的字符串，对这个字符串的格式来进行解释：
- dialect+driver://username:password@host:port/database
- dialect是数据库的实现，比如MySQL、PostgreSQL、SQLite，并且转换成小写。driver是Python对应的驱动，如果不指定，会选择默认的驱动，比如MySQL的默认驱动是MySQLdb。username是连接数据库的用户名，password是连接数据库的密码，host是连接数据库的域名，port是数据库监听的端口号，database是连接哪个数据库的名字。
- 如果以上输出了1，说明SQLAlchemy能成功连接到数据库。
SQLAlchemy参数
- Column常用参数：
- default：默认值。
- nullable：是否可空。
- primary_key：是否为主键。
- unique：是否唯一。
- autoincrement：是否自动增长。
- onupdate：更新的时候执行的函数。
- name：该属性在数据库中的字段映射。
- sqlalchemy常用数据类型：
- Integer：整形。
- Float：浮点类型。
- Boolean：传递True/False进去。
- DECIMAL：定点类型。
- enum：枚举类型。
- Date：传递datetime.date()进去。
- DateTime：传递datetime.datetime()进去。
- Time：传递datetime.time()进去。
- String：字符类型，使用时需要指定长度，区别于Text类型。
- Text：文本类型。
- LONGTEXT：长文本类型。
- query可用参数：
- 模型对象。指定查找这个模型中所有的对象。
- 模型中的属性。可以指定只查找某个模型的其中几个属性。
- 聚合函数。
- func.count：统计行的数量。
- func.avg：求平均值。
- func.max：求最大值。
- func.min：求最小值。
- func.sum：求和。
创建表
- # -*- coding: utf-8 -*-
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy import Column # 列
- from sqlalchemy import Integer, String # 属性
- Base = declarative_base() # django models
- # 创建表
- class User(Base):
- __tablename__ = 'user'
- id = Column(Integer, primary_key=True, autoincrement=True)
- name = Column(String(32), index=True, name='名字')
- # 数据库连接
- from sqlalchemy import create_engine
- engine = create_engine("mysql+pymysql://root:redhat@192.168.32.71:3306/my_sql?charset=utf8")
- # 去engine数据库中创建所有继承Base的表
- Base.metadata.create_all(engine)
增加数据
- # -*- coding: utf-8 -*-
- from .create import engine, User
- # 增加数据
- # 创建会话窗口
- from sqlalchemy.orm import sessionmaker
- Session = sessionmaker(engine)
- # 打开会话窗口
- db_session = Session()
- # ---单条数据
- user_obj = User(name='Ywb') # 实例化
- db_session.add(user_obj) # 相当于 insert into
- # 执行会化窗口中的所有操作
- db_session.commit()
- db_session.close()
- # ---增加批量数据
- db_session.add_all([
- User(name='peach'),
- User(name='小红')
- ])
- db_session.commit()
- db_session.close()
- # ---扩展
- user1 = User(name='11')
- user2 = User(name='12')
- user3 = User(name='13')
- user4 = User(name='14')
- db_session.add(user1)
- db_session.add(user2)
- db_session.add(user3)
- db_session.add(user4)
- db_session.commit() # 全部添加进去
- db_session.close()
过滤
- 过滤是数据提取的一个很重要的功能，以下对一些常用的过滤条件进行解释，并且这些过滤条件都是只能通过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']))
- # 同时，in也可以作用于一个Query
- 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)
- # 或者是
- query.filter(User.name.is_(None))
- is not null:
- query.filter(User.name != None)
- # 或者是
- query.filter(User.name.isnot(None))
- and：
- from sqlalchemy import and_
- query.filter(and_(User.name=='ed',User.fullname=='Ed Jones'))
- # 或者是传递多个参数
- query.filter(User.name=='ed',User.fullname=='Ed Jones')
- # 或者是通过多次filter操作
- query.filter(User.name=='ed').filter(User.fullname=='Ed Jones')
- or：
- from sqlalchemy import or_
- query.filter(or_(User.name=='ed',User.name=='wendy'))
单表查询
- # -*- coding: utf-8 -*-
- # 单表查询
- from sqlalchemy.orm import sessionmaker
- from SQLAlchemy.create import engine, User
- Session = sessionmaker(engine)
- db_session = Session()
- # ---基本查询
- # select * from name
- user_list = db_session.query(User)
- print(user_list) # SELECT user.`名字` AS `user_名字`, user.id AS user_id FROM user
- # ---------------- 所有数据 ------------------
- user_list = db_session.query(User).all() # [obj, obj]
- for usr in user_list:
- print(usr.name)
- # ---------------- 一条数据 ------------------
- user = db_session.query(User).first() # obj
- print(user.name)
- # ---------------- 带条件查询 ------------------
- # --- filter
- user_list = db_session.query(User).filter(User.id==2).all() # [obj, obj]
- print(user_list)
- user_list = db_session.query(User).filter(User.id >= 1).all() # [obj, obj]
- for user in user_list:
- print(user.id) # 按照id排序，因为添加是id
- # --- filter_by
- user = db_session.query(User).filter_by(id=2).first() # obj
- print(user)
- # --- 扩展-查看sql语句
- sql = db_session.query(User).filter(User.id >= 1)
- print(sql)
- """
- SELECT user.`名字` AS `user_名字`, user.id AS user_id
- FROM user
- WHERE user.id >= %(id_1)s
- """
修改
- # -*- coding: utf-8 -*-
- # 更新修改数据
- from sqlalchemy.orm import sessionmaker
- from SQLAlchemy.create import engine,User
- Session = sessionmaker(engine)
- db_session = Session()
- # 单条修改
- # update `uesr` set `name` = 'haha' where id =1
- res = db_session.query(User).filter(User.id==1).update({
- "name": 'haha'
- })
- print(res) # 返回影响的行数
- db_session.commit()
- db_session.close()
- # 批量修改
- res = db_session.query(User).filter(User.id > 1).update({
- "name": "1234"
- })
- print(res)
- db_session.commit()
- db_session.close()
删除
- # -*- coding: utf-8 -*-
- # 删除数据
- from sqlalchemy.orm import sessionmaker
- from SQLAlchemy.create import engine,User
- Session = sessionmaker(engine)
- db_session = Session()
- # 删除单条
- # delete from user where id = 1
- res = db_session.query(User).filter(User.id==1).delete()
- print(res) # 返回影响的行数
- db_session.commit()
- db_session.close()
- # 删除多条
- res = db_session.query(User).filter(User.name=="12345").delete()
- print(res)
- db_session.commit()
- db_session.close()