对数据库表的增
models.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import Column,Integer,String,Text,ForeignKey,DateTime,UniqueConstraint,Index
Base = declarative_base()
class Users(Base):
__tablename__ ="users"
id = Column(Integer,primary_key=True)
name = Column(String(32),index=True,nullable=False,default="0")
age = Column(Integer)
#email = Column(String(32),unique=True)
def __repr__(self):
return self.name
def init_db():
'''
通过类来创建表
'''
engine = create_engine(#创建引擎
#"mysql+pymysql://用户名：密码@ip:端口/库名？charset=字符集"
# 1 用户名密码错误，用户权限不够，
# 2 ip和端口不对。远程ip,该服务器的防火墙有没有关闭，端口不一定是3306，3307，3308，3309，3310
"mysql+pymysql://root:@127.0.0.1:3306/python12?charset=utf8",
max_overflow = 0,
pool_size =5,
pool_timeout = 30,
pool_recycle = -1
)
Base.metadata.create_all(engine)
def del_db():
'''
根据类来删除
'''
engine = create_engine(
"mysql+pymysql://root:@127.0.0.1:3306/python12?charset=utf8",
max_overflow=0,
pool_size=5,
pool_timeout=30,
pool_recycle=-1
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
# sqlalchemy是不支持改表的。
init_db()
#del_db()
增.py
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#创建引擎
engine=create_engine("mysql+pymysql://root:@127.0.0.1:3306/python12", max_overflow=0, pool_size=5)
# 绑定引擎 连接
Connection=sessionmaker(bind=engine)
#每次执行数据库操作是，都遇到创建一个Connection对象
con=Connection()
################执行ORM操作######################
obj1=Users(name="ojbk")
con.add(obj1)
#提交事务
con.commit()
#关闭session 其实是将连接放回连接池
con.close()
secop线程安全
secop线程安全.py
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Users
#创建引擎
engine=create_engine("mysql+pymysql://root:@127.0.0.1:3306/python12", max_overflow=0, pool_size=5)
# 绑定引擎 连接
Session=sessionmaker(bind=engine)
'''
线程安全，基于本地线程实现每隔额线程用同一个session
特殊性：scoped_session中有原来方法的Session的所有方法
public_methods = (
'__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',
'close', 'commit', 'connection', 'delete', 'execute', 'expire',
'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind',
'is_modified', 'bulk_save_objects', 'bulk_insert_mappings',
'bulk_update_mappings',
'merge', 'query', 'refresh', 'rollback',
'scalar'
)
'''
#scoped_session类并没有继承Session 但是却有他的所有方法
session=scoped_session(Session)
###################执行ORM操作#################################
obj1=Users(name="asdd")
session.add(obj1)
#提交事务
session.commit()
#关闭session
session.close()
单表的增删改查
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#创建引擎
engine=create_engine("mysql+pymysql://root:@127.0.0.1:3306/python12", max_overflow=0, pool_size=5)
# 绑定引擎 连接
Session=sessionmaker(bind=engine)
#每次执行数据库操作是，都遇到创建一个Connection对象
session=Session()
# 增
#方式一
'''
obj1=Users(name="tank")
session.add(obj1)
'''
# 方式二
'''
session.add_all([
Users(name="jsaon"),
Users(name="egon"),
#order(monery=10)
])
'''
#删
'''session.query(Users).filter(Users.id==1).delete()
'''
#改
'''
session.query(Users).filter(Users.id==2).update({Users.name:"jdkdkdkdk"})
session.query(Users).filter(Users.id==2).update({"name":"kdkdk"})
'''
'''
类似于django中的F查询，但是必须要配合synchronize_session使用
如果是字符串 就用synchronize_session=False
如果是数字 就用synchronize_session="evaluate"，可以不写 默认就是这个
'''
# session.query(Users).filter(Users.id==2).update({"name":Users.name+"hahah"},synchronize_session=False)
# session.query(Users).filter(Users.id==2).update({Users.age:Users.age+123},synchronize_session="evaluate")
# 所有的增删改都需要session.commit()
'''
session.commit()
session.close()
'''
#查
#打印SQL
# r1=session.query(Users)
#获取所有数据
# r1=session.query(Users).all()
#筛选，有两种 一种是filter 一种是filter_by
#filter :传递的是表达式
# filter_by ：传递的是参数
r1=session.query(Users).filter(Users.id>3).all()
r2=session.query(Users).filter_by(name="tank").all()
print(r1)
print(r2)
单表的高级查询
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#创建引擎
engine=create_engine("mysql+pymysql://root:@127.0.0.1:3306/python12", max_overflow=0, pool_size=5)
# 绑定引擎 连接
Session=sessionmaker(bind=engine)
#每次执行数据库操作是，都遇到创建一个Connection对象
session=Session()
# 条件查询
ret=session.query(Users).filter_by(name="tank").all()
# 表达式 and条件连接
ret=session.query(Users).filter(Users.name=="jason",Users.age>2).all()
#between查询
ret=session.query(Users).filter(Users.id.between(3,5)).all()
# in查询
ret=session.query(Users).filter(Users.id.in_([3,4,5,6])).all()
# 取反
ret=session.query(Users).filter(~Users.id.in_([3,4,5,6])).all()
# or与and
from sqlalchemy import and_,or_
# ret=session.query(Users).filter(and_(Users.id>5,Users.name=="jason")).all()
ret=session.query(Users).filter(
and_(
or_(Users.name=="jason",Users.name=="tank"),
Users.id==6
)
).all()
#like查询
ret=session.query(Users).filter(Users.name.like("j%")).all()#以j开头
ret=session.query(Users).filter(Users.name.like("_a%")).all()#第二个是a
#排序
#从大到小
ret=session.query(Users).order_by(Users.id.desc()).all()
#从小到大
ret=session.query(Users).order_by(Users.id.asc()).all()
#多标准排序
ret=session.query(Users).order_by(Users.age.asc(),Users.id.desc()).all()
#分组
ret=session.query(Users).group_by(Users.age).all()
from sqlalchemy.sql import func
ret=session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)
).group_by(Users.age).all()
#二次条件
ret=session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)
).group_by(Users.age).having(func.min(Users.id)>5).all()
print(ret)
一对多
models.py
#一个爱好，可以被多个人喜欢。
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
class Person(Base):
__tablename__ = 'person'
nid = Column(Integer, primary_key=True)
name = Column(String(32), nullable=True)
# hobby指的是tablename而不是类名，uselist=False 外键
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关，不会新增字段，只用于快速链表操作 类似于虚拟字段
# 类名，backref用于反向查询
hobby = relationship('Hobby', backref='pers')
一对多.py
import time
import threading
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
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from models import Users,Hobby, Person
#创建引擎
engine=create_engine("mysql+pymysql://root:@127.0.0.1:3306/python12", max_overflow=0, pool_size=5)
# 绑定引擎 连接
Session=sessionmaker(bind=engine)
#每次执行数据库操作是，都遇到创建一个Connection对象
session=Session()
# 增
# session.add_all([
# Hobby(caption="洗澡"),
# Hobby(caption="洗脚"),
# Person(name="甲",hobby_id=1),
# Person(name="乙",hobby_id=2)
# ])
# per=Person(name="丙",hobby=Hobby(caption="搓背"))
# per=Person(name="丙",hobby_id=1)
# session.add(per)
# ho=Hobby(caption="按摩")
# ho.pers=[Person(name="丁"),Person(name="戊")]
# session.add(ho)
# #
# session.commit()
# session.close()
# 查
person_list=session.query(Person).all()
print(person_list[0].name,person_list[0].hobby.caption)
hob=session.query(Hobby).filter(Hobby.id==5).first()
print(hob.id)
print(hob.pers[0].name)
多对多
models.py
class Boy2Girl(Base):
__tablename__ = 'boy2girl'
id = Column(Integer, primary_key=True, autoincrement=True)
girl_id = Column(Integer, ForeignKey('girl.id'))
boy_id = Column(Integer, ForeignKey('boy.id'))
class Girl(Base):
__tablename__ = 'girl'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
class Boy(Base):
__tablename__ = 'boy'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
girl = relationship('Girl', secondary='boy2girl', backref='boys')
多对多.py
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Girl,Boy,Boy2Girl
engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/python12?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 增
# session.add_all([
# Girl(name="苍"),
# Girl(name="玛"),
# Boy(name="甲"),
# Boy(name="乙")
# ])
# b2g=Boy2Girl(girl_id=1,boy_id=1)
# session.add(b2g)
# 关联增加
# boy1=Boy(name="丁")
# boy1.girl=[Girl(name="风"),Girl(name="真")]
# session.add(boy1)
# gir=Girl(name="郭")
# gir.boys=[Boy(name="李易峰"),Boy(name="彭于晏")]
# session.add(gir)
# session.commit()
# session.close()
#正向查询
boys=session.query(Boy).first()
print(boys.name)
print(boys.girl[0].name)
girls=session.query(Girl).filter(Girl.id==6).first()
print(girls.name)
print(girls.boys[0].name,girls.boys[1].name)
pro_flask_sansa
pip install flask-sqlalchemy
pip install flask-migrate
python manage.py db init 初始化：只执行一次
python manage.py db migrate 等同于 makemigartions
python manage.py db upgrade 等同于migrate
-sansa
--views
---account.py
--__init__.py
--models.py
-manage.py
-settings.py
account.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Blueprint
from .. import db
from .. import models
account = Blueprint('account', __name__)
@account.route('/login')
def login():
db.session.add(models.Users(username='lqz', email='123'))
db.session.commit()
tt = db.session.query(models.Users).all()
print(tt[0].username)
# 添加示例
"""
db.session.add(models.Users(username='lqz', pwd='123', gender=1))
db.session.commit()
obj = db.session.query(models.Users).filter(models.Users.id == 1).first()
print(obj)
PS: db.session和db.create_session
"""
# db.session.add(models.Users(username='wupeiqi1', email='wupeiqi1@xx.com'))
# db.session.commit()
# db.session.close()
#
# db.session.add(models.Users(username='wupeiqi2', email='wupeiqi2@xx.com'))
# db.session.commit()
# db.session.close()
# db.session.add(models.Users(username='alex1',email='alex1@live.com'))
# db.session.commit()
# db.session.close()
# user_list = db.session.query(models.Users).all()
# db.session.close()
# for item in user_list:
# print(item.username)
return 'login'
init.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
from .models import *
from .views import account
def create_app():
app = Flask(__name__)
app.config.from_object('settings.DevelopmentConfig')
# 将db注册到app中
db.init_app(app)
# 注册蓝图
app.register_blueprint(account.account)
return app
models.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from . import db
class Users(db.Model):
"""
用户表
"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
# ids = db.Column(db.Integer)
def __repr__(self):
return '<User %r>' % self.username
manage.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
生成依赖文件：
pipreqs ./
"""
from sansa import create_app
from flask_script import Manager
from flask_migrate import Migrate,MigrateCommand
from sansa import db
app = create_app()
manager=Manager(app)
Migrate(app,db)
manager.add_command('db', MigrateCommand)
if __name__ == '__main__':
# app.run()
manager.run()
settings.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
class BaseConfig(object):
# SESSION_TYPE = 'redis' # session类型为redis
# SESSION_KEY_PREFIX = 'session:' # 保存到session中的值的前缀
# SESSION_PERMANENT = True # 如果设置为False，则关闭浏览器session就失效。
# SESSION_USE_SIGNER = False # 是否对发送到浏览器上 session:cookie值进行加密
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:@127.0.0.1:3306/python12?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False
class ProductionConfig(BaseConfig):
pass
class DevelopmentConfig(BaseConfig):
pass
class TestingConfig(BaseConfig):
pass