数据库和Flask项目搭建

news/2024/7/7 7:44:58/文章来源:https://www.cnblogs.com/banchengyanyu/p/18272051

1 sqlalchemy原生操作

# 操作原生sql ---》用得少
import pymysql
import threading
# 1 导入
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
# 2 创建引擎
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/cnblogs",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,     # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)# 3 使用引擎，拿到链接
# conn = engine.raw_connection()
# # 4 剩下的一样了
# cursor=conn.cursor(pymysql.cursors.DictCursor)
# cursor.execute('select * from article limit 10')
# res=cursor.fetchall()
#
# print(res)## 多线程测试：
def task(arg):conn = engine.raw_connection()cursor = conn.cursor()cursor.execute("select * from article")result = cursor.fetchall()print(result)cursor.close()conn.close()for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()

2 sqlalchemy操作表

# 1 建个 models.py  里面写 表模型
# 2 把表模型---》同步到数据库中
# 3 增删查改

2.1 创建删除表

# 1 导入
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, IndexBase = declarative_base()  # Base 当成 models.Model# 2 创建表模型
class User(Base):__tablename__ = 'users'  # 表名# 写字段id = Column(Integer, primary_key=True, autoincrement=True)  # id 主键name = Column(String(32), index=True, nullable=False)  # name列，索引，不可为空email = Column(String(32), unique=True)# datetime.datetime.now不能加括号，加了括号，以后永远是当前时间ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text)# 3 没有命令---》后期使用第三方模块，可以有命令
# 目前需要手动做
# sqlalchemy 不能创建数据库，能创建表，删除表，不能删除增加字段（第三方模块）
# 3.1 创建引擎
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy01",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
## 3.2 把表模型同步到数据库中
Base.metadata.create_all(engine)# 3.3 删除表
# Base.metadata.drop_all(engine)

2.2 简单的增删改查

from models import Userfrom sqlalchemy import create_engine# 1 创建引擎
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy01",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)# 2 orm操作--》借助于 engine 得到session（conn）对象
from sqlalchemy.orm import sessionmakerConnection = sessionmaker(bind=engine)
conn = Connection()
# 3 使用conn---》进行orm操作
# 3.1 增加数据
# user = User(name='lqz', email='3@qq.com')
# # 插入到数据库
# conn.add(user)  # 放个对象
# # 提交
# conn.commit()
# # 关闭链接
# conn.close()# 3.2 查询数据
# 查询User表中id为1的所有记录--》放到列表中
# res=conn.query(User).filter_by(id=1).all()
# print(res)## 3.3 删除
# res = conn.query(User).filter_by(name='lqz').delete()
# print(res)
# conn.commit()## 3.4 修改# res=conn.query(User).filter_by(name='9999').update({'extra':'xxsss'})
# conn.commit()

3 一对多关系

# 1 一对一   [本质就是一对多--》多的那个唯一]
# 2 一对多
# 3 多对多

3.1 关系

####  一对多关系
class Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)caption = Column(String(50), default='篮球')def __str__(self):return self.captionclass Person(Base):__tablename__ = 'person'nid = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)# hobby指的是tablename而不是类名hobby_id = Column(Integer, ForeignKey("hobby.id"))  # 一个爱好，可以被多个人喜欢，一个人只能喜欢一个爱好# 跟数据库无关，不会新增字段，只用于快速链表操作# 类名，backref用于反向查询hobby = relationship('Hobby', backref='pers')def __str__(self):return self.namedef __repr__(self):return self.name

from models import Person,Hobby
# 第一步：创建engine
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3307/sqlalchemy001?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)# 第二步：创建 session对象---老方式
from sqlalchemy.orm import Session
session=Session(engine)if __name__ == '__main__':### 笨办法增加#1 先增加一个hobby# hobby=Hobby()# session.add(hobby)# session.commit()#2 增加Person---》必须要有hobby_id# person=Person(name='lqz',hobby_id=1)# session.add(person)# session.commit()### 简便方法--》增加person的同时，增加了Hobby# person = Person(name='hope123', hobby=Hobby(caption='乒乓球'))# session.add(person)# session.commit()# hobby=session.query(Hobby).filter_by(id=1).first()# person = Person(name='hope', hobby=hobby)# session.add(person)# session.commit()# 基于对象的跨表查询--->正向# person=session.query(Person).filter_by(nid=2).first()# print(person)# print(person.hobby_id)# print(person.hobby)# 基于对象的跨表查询--->反向hobby = session.query(Hobby).filter_by(id=1).first()print(hobby.caption)print(hobby.pers)

4 scoped线程安全

# 1 如果我们把session 做成全局 单例-每个视图函数，用同一个session，有问题#2 如果在每个视图函数中，都对session实例化一次-代码有点麻烦# 3 全局就用一个session对象，它在不同线程中---》都是这个线程自己的

from models import User
import threading
# 第一步：创建engine
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy001?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)# 第二步：创建 线程安全的 session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
Session = sessionmaker(bind=engine)
session=scoped_session(Session)# 第三步：正常使用-->再flask中，使用全局的session即可，实现：不同线程使用线程自己的session对象
def task(se,i):session=se()session.add(User(name='xxx',email=f'{i}@qq.com'))session.commit()print('=========',session)if __name__ == '__main__':l=[]for i in range(10):t=threading.Thread(target=task,args=[session,i])t.start()l.append(t)for i in l:i.join()

5多对多关系

5.1 表模型

#### 多对多
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'))ctime = Column(DateTime, default=datetime.datetime.now)class Girl(Base):__tablename__ = 'girl'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)def __repr__(self):return self.nameclass Boy(Base):__tablename__ = 'boy'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(64), unique=True, nullable=False)# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以girls = relationship('Girl', secondary='boy2girl', backref='boys')def __repr__(self):return self.name

5.2 操作

from models import Boy,Girl,Boy2Girl
# 第一步：创建engine
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy001?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)# 第二步：创建 session对象---老方式
from sqlalchemy.orm import Session
session=Session(engine)if __name__ == '__main__':### 1 增加记录--笨办法# boy=Boy(name='张四')# boy1=Boy(name='张五')# girl=Girl(name='小粉')# girl1=Girl(name='小紫')# session.add_all([boy,girl,boy1,girl1])# session.commit()## 2 往中间表增加记录# res=Boy2Girl(boy_id=1,girl_id=1)# session.add_all([res])# session.commit()## 3 简便方式增加记录girls=session.query(Girl).filter(Girl.id>=2).all()print(girls)## 3.1 让id为2的张四，跟上面俩女生约会# boy = session.query(Boy).filter(Boy.id == 2).all()[0]# print(boy)# boy.girls=girls# session.add(boy)# session.commit()## 3.2 新用户跟上面俩女生约会# boy = Boy(name='彭于晏',girls=girls)# session.add(boy)# session.commit()# 4 基于对象的跨表查--》正向# boy = session.query(Boy).filter(Boy.id == 2).all()[0]# print(boy.name)# print(boy.girls)girl = session.query(Girl).filter(Girl.id == 2).all()[0]print(girl.name)print(girl.boys)

6 常用查询筛选

from models import User,Person,Hobby,Boy,Girl,Boy2Girl
# 第一步：创建engine
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3307/sqlalchemy001?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)# 第二步：创建 session对象---老方式
from sqlalchemy.orm import Session
session=Session(engine)if __name__ == '__main__':## 1 filter_by查询  写参数# res=session.query(User).filter_by(name='hope11',id=4).all()# res=session.query(User).filter_by(name='hope11').all() # 返回列表# res=session.query(User).filter_by(name='hope11').first() # 返回单条，但是不能all() 在用first，因为all后是列表## 2 filter[where]查询  写条件# res = session.query(User).where(User.id>=1).all()  # 就是filter# res = session.query(User).filter(User.id>=1).all()# res = session.query(User).filter(User.id==2).all()# res = session.query(User).filter(User.id>=2) # 原生sql## 3 between# res = session.query(User).where(User.id.between(1, 38), User.name == 'xxx').all()## 4 in# res = session.query(User).filter(User.id.in_([1, 3, 4,37])).all()## 5 ~非，除。。外# res = session.query(User).filter(~User.id.in_([1, 3, 4]),User.name=='xxx').all()# 6 二次筛选# select * from user where user.id in (select id from user where name=xxx)# res = session.query(User).filter(User.id.in_(session.query(Person.nid).filter_by(name='lqz'))).all()# 7 and,or条件from sqlalchemy import and_, or_# or_包裹的都是or条件，and_包裹的都是and条件# res = session.query(User).filter(and_(User.id > 3, User.name == 'eric')).all()# res = session.query(User).filter(or_(User.id < 2, User.name == 'eric')).all()# res = session.query(User).filter(#     or_(#         User.id < 2,#         and_(User.name == 'eric', User.id > 3),#         User.extra != ""#     ))# print(res)# 8 通配符，以e开头，不以e开头# res = session.query(User).filter(User.name.like('l%')).all()# ret = session.query(User).filter(~User.name.like('l%'))# 9 限制，用于分页，区间# ret = session.query(User)[0:3]  # 第一页，一页显示三条# 10 排序，根据name降序排列（从大到小） desc   asc# ret = session.query(User).order_by(User.name.asc()).all()# 11 第一个条件重复后，再按第二个条件升序排   字符串按什么排？# ret = session.query(User).order_by(User.name.desc(), User.id.asc()).all()# 12 分组      一旦分组后，select 字段必须是 分组字段或聚合函数的字段from sqlalchemy.sql import func# select name from user group by name# ret = session.query(User).group_by(User.name).all() # 分组完要统计--》这个sql一般不写#  分组之后取最大id，id之和，最小id# select max(id),sum(id),min(id),name from user group by user.name;# ret = session.query(#     func.max(User.id),#     func.sum(User.id),#     func.min(User.id),User.name).group_by(User.name).all()# # haviing筛选# select max(id),sum(id),min(id),name from user where id >=4 group by user.name;# User.objects.values('name').filter(id__gte=4).anoto(m=max('id'),s=sum('id'),i=min('id')).filter(s__gt=400.values(m,s,i,name)# ret = session.query(#     func.max(User.id),#     func.sum(User.id),#     func.min(User.id),User.name).filter(User.id>=4).group_by(User.name).having(func.sum(User.id) > 400).all()# 13 连表（默认用forinkey关联）## 笛卡尔积链表# select * from person,hobby where person.hobby_id=hobby.id;# ret = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all() #笛卡尔积后过滤#  join表，默认是inner join 没有指定on字段，默认按外键# select * from person INNER join hobby on person.hobby_id=hobby.id;# ret = session.query(Person).join(Hobby)#  isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可# select * from person left join hobby on person.hobby_id=hobby.id;# ret = session.query(Person).join(Hobby, isouter=True)# ret = session.query(Hobby).join(Person, isouter=True)# # 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上   因为我们可以没有强外键关联ret = session.query(Person).join(Hobby, Person.nid == Hobby.id, isouter=True)########## 留着 ############# # 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集# # union和union all的区别？# q1 = session.query(Users.name).filter(Users.id > 2)# q2 = session.query(Favor.caption).filter(Favor.nid < 2)# ret = q1.union(q2).all()## q1 = session.query(Users.name).filter(Users.id > 2)# q2 = session.query(Favor.caption).filter(Favor.nid < 2)# ret = q1.union_all(q2).all()print(ret)

7 flask-sqlalchemy使用

7.1 项目目录

7.2 使用flask-sqlalchemy集成到flask中

安装

pip install flask_sqlalchemy

init

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
app = Flask(__name__, template_folder='../templates', static_folder='../static')
app.config.from_pyfile('./settings.py')db.init_app(app)
from .views import bp_userapp.register_blueprint(bp_user)

models

import datetimefrom . import db### 单表
class User(db.Model):__tablename__ = 'users'  # 表名# 写字段id = db.Column(db.Integer, primary_key=True, autoincrement=True)  # id 主键name = db.Column(db.String(32), index=True, nullable=False)  # name列，索引，不可为空email = db.Column(db.String(32), unique=True)# datetime.datetime.now不能加括号，加了括号，以后永远是当前时间ctime = db.Column(db.DateTime, default=datetime.datetime.now)extra = db.Column(db.Text)def __str__(self):return self.namedef __repr__(self):return self.name

settings

DEBUG=True
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy003?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False

views

from flask import Blueprint, render_template, jsonify
from .models import User# bp_user = Blueprint('user', __name__)
#url_prefix可以为增加url的path部分的前缀，可以更方便的去管理访问视图函数。
user_bp = Blueprint('user',__name__,url_prefix='/user')
from . import db@bp_user.route('/')
def index():user=User(name='xxx',email='55@qq.com')db.session.add(user)db.session.commit()# 把用户表中所有数据都查出来，返回给前端res = db.session.query(User).filter(User.id >= 1).all()# 不行 json序列化，只能序列化 数字，字符串，布尔，列表，和字典# res 列表套对象---》不能序列化# l = []# for item in res:#     l.append({'name': item.name, 'email': item.email})# return jsonify({'code': 100, 'msg': "成功", 'results': l})return render_template('index.html',users=res)

8 flask-migrate使用

# 1 数据库肯定要自己创建
# 2 创建表，增加删除字段---》手动做---》django 有两个命令---》自动做-有没有种方案，跟djagno一样，自动记录，自动迁移# 3 django中：python manage.py makemigrationspython manage.py migrate 
# 4 第三方模块：flask-migrate--》完成跟django一样--》命令稍微有些不同# https://github.com/miguelgrinberg/Flask-Migrate/
pip3 install Flask-Migrate --upgrade
4.0.7####### 使用
from flask_migrate import Migrateapp = Flask(__name__)
app.config.from_pyfile('./settings.py')
db = SQLAlchemy(app)
# db.init_app(app)
migrate = Migrate(app, db) # flask 就会多出好几个命令---》# flask --app manage:app db init  # 初始化，第一次执行，以后再也不执行了，它执行完，会出现一个migrations文件夹
# flask --app manage:app db migrate # django中的makemigrations 是一模一样
# flask --app manage:app db upgrade  # 跟django的migrate一样
# flask上其他第三方插件corstokencacherestful

9 flask 项目

9.1 项目目录

#  网上开源的---》前后端混合layui---》rbac的权限控制下载下来---》运行---》研究功能flask db initflask db migrate -m '数据初始化'flask db upgradeflask admin init# https://gitee.com/pear-admin/pear-admin-flask

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/732330.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

flask定制Excel命令

定制excel命令并插入到数据库中 import osfrom flask import Flask from flask.cli import AppGroup import click import pymysql from openpyxl import load_workbookapp = Flask(__name__)@app.cli.group() def excel():"""Excel related commands."&qu…