|
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()
|
