一 。 flask 中连接的数据库的组件 DButils
#!/usr/bin/env python # -*- coding:utf-8 -*- import pymysql from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB( creator=pymysql, # 使用链接数据库的模块 maxconnections=20, # 连接池允许的最大连接数,0和None表示不限制连接数 mincached=2, # 初始化时,链接池中至少创建的空闲的链接,0表示不创建 maxcached=5, # 链接池中最多闲置的链接,0和None不限制 maxshared=0, # 链接池中最多共享的链接数量,0和None表示全部共享。PS: 无用,因为pymysql和MySQLdb等模块的 threadsafety都为1,所有值无论设置为多少,_maxcached永远为0,所以永远是所有链接都共享。 blocking=True, # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错 maxusage=None, # 一个链接最多被重复使用的次数,None表示无限制 setsession=[], # 开始会话前执行的命令列表。如:["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服务端,检查是否服务可用。# 如:0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always host=‘127.0.0.1‘, port=3306, user=‘root‘, password=‘123456‘, database=‘day116‘, charset=‘utf8‘ ) def func(): # 去连接池中获取一个连接 conn = POOL.connection() cursor = conn.cursor() # cursor.execute(‘select * from users‘) print(‘开始去执行了‘) cursor.execute(‘select sleep(5)‘) result = cursor.fetchall() # 将连接返还到连接池中 conn.close() # print(result) def func2(): conn = POOL.connection() cursor = conn.cursor() # cursor.execute(‘select * from users‘) print(‘开始去执行了‘,2,3) cursor.execute(‘select sleep(5)‘) result = cursor.fetchall() # 将连接返还到连接池中 conn.close() # print(result) import threading # 多线程的模块 for i in range(20): t = threading.Thread(target=func) t2 = threading.Thread(target=func2) t.start() t2.start()
二 。 flask 中的 flask-session 组件
将Flask中的session由原来的默认写到加密cookie中,改成放置到其他数据源,如:redis/memcached/filesystem/mongodb/sqlalchemy(数据库
在 setttings 中设置的代码
SECRET_KEY = "asdfasdfasdf" SESSION_TYPE = ‘redis‘ SESSION_REDIS = Redis(host=‘127.0.0.1‘,port=6379)
在 app 中的代码
from flask import Flask,session,request,redirect from flask_session import Session # 导入这个模块 from .views.account import account from .views.code import code from .views.student import student def auth(): """ 用户认证 :return: """ if request.path == ‘/login‘: return None user_info = session.get(‘user_info‘) if user_info: return None return redirect(‘/login‘) def create_app(): app = Flask(__name__) app.config.from_object(‘settings.DevelopmentConfig‘) app.register_blueprint(account) app.register_blueprint(code) app.register_blueprint(student) app.before_request(auth) Session(app) # 把app对象传进去 return app
三 flask 中的 wtforms 组件:
跟Django中的form组件 一样 都是用于做表单验证
from wtforms.fields import simple from wtforms.fields import html5 from wtforms.fields import core from wtforms import widgets from wtforms import validators from wtforms import Form from zzy.utils import sqlhelper class ClsForm(Form): title = simple.StringField( label=‘名称‘, widget=widgets.TextInput(), render_kw={‘class‘: ‘form-control‘}, validators=[ validators.DataRequired(message=‘班级名称不能为空‘) ], ) class TestForm(Form): name = simple.StringField( label=‘用户名‘, validators=[ validators.DataRequired() ], widget=widgets.TextInput(), render_kw={‘class‘: ‘form-control‘} ) """ pwd = simple.PasswordField( label=‘密码‘, validators=[ validators.DataRequired(message=‘密码不能为空.‘) ], widget=widgets.PasswordInput(), render_kw={‘class‘: ‘form-control‘} ) pwd_confirm = simple.PasswordField( label=‘重复密码‘, validators=[ validators.DataRequired(message=‘重复密码不能为空.‘), validators.EqualTo(‘pwd‘, message="两次密码输入不一致") ], widget=widgets.PasswordInput(), render_kw={‘class‘: ‘form-control‘} ) email = html5.EmailField( label=‘邮箱‘, validators=[ validators.DataRequired(message=‘邮箱不能为空.‘), validators.Email(message=‘邮箱格式错误‘) ], widget=widgets.TextInput(input_type=‘email‘), render_kw={‘class‘: ‘form-control‘} ) gender = core.RadioField( label=‘性别‘, choices=( (1, ‘男‘), (2, ‘女‘), ), coerce=int ) city = core.SelectField( label=‘城市‘, choices=( (‘bj‘, ‘北京‘), (‘sh‘, ‘上海‘), ) ) hobby = core.SelectMultipleField( label=‘爱好‘, choices=( (1, ‘篮球‘), (2, ‘足球‘), ), coerce=int ) favor = core.SelectMultipleField( label=‘喜好‘, choices=( (1, ‘篮球‘), (2, ‘足球‘), ), widget=widgets.ListWidget(prefix_label=False), option_widget=widgets.CheckboxInput(), coerce=int ) """ cls_id = core.SelectField( label=‘请选择班级‘, # choices=( # (‘bj‘, ‘北京‘), # (‘sh‘, ‘上海‘), # ) choices=[], coerce=int ) def __init__(self, *args, **kwargs): super(TestForm, self).__init__(*args, **kwargs) self.cls_id.choices = sqlhelper.fetchall(sql=‘select id,title from classes‘,args=[],cur=None) def validate_name(self,field): """ 对name进行验证时的钩子函数 :param field: :return: """ if field != ‘root‘: raise validators.ValidationError("用户名必须是root")
不限语言的
四 pipreqs 组件:
安装: pip3 install pipreqs 使用: pipreqs ./ --encoding=utf-8 # 执行完生产一个 requirements.txt 文件 ,里面记录了此项目的所有模块和版本, pip3 install -r requirements.txt 可以下载文件里的所有模块
不限语言的
五:virtualenv
创建虚拟的环境
安装: pip3 install virtualenv 命令创建虚拟环境: virtualenv 虚拟环境的名称 cd 虚拟环境目录 activate 激活虚拟环境 pip3 install django==1.7 deactivate 退出虚拟环境
原文地址:https://www.cnblogs.com/xuerh/p/9373732.html
时间: 2024-09-27 06:20:16