Flask

靓盈 · 发表于 2023-7-18 15:38:03

Flask

短小精悍，可扩展强的一个web框架。上下文管理
安装：pip3 install Flask，我们做web开发时，是站在两大东西之上做的web框架和wsgi，Flask和Django中都是应用的并不是自己写的。Flask中werkzurg就是。
web服务网关接口，wsgi是一个协议，实现该写一个的模块：wsgiref|werkzeug，实现其协议的模块本质上就是socket服务端用于接收用户请求，并处理。一般web框架基于wsgi实现，这样实现关注点分离。
flask就是基于Werkzurg搭建起来的
1. from werkzeug.wrappers import Request, Response
2. from werkzeug.serving import run_simple
4. @Request.application
5. def run(environ, start_response):
6. return Response("Hello World!") # 返回字符串
7. if __name__=="__main__":
8. run_simple("localhost", 4000, run) # 监听本机的4000端口，如果访问就执行run函数
复制代码
一个最简单的Flask
1. from flask import Flask
2. app = Flask(__name__) # flask名称，一般这样写。
4. @app.route("/index") # url路径
5. def index(): #所调用的函数
6. return "Hello World!"
7. if __name__ == "__main__":
8. app.run() # 执行口
复制代码
Flask实现用户登录程序
1. from flask import Flask, render_template, request, redirect, session
2. app = Flask(__name__, template_folder="templates", static_folder="static")
3. app.secret_key = "asdf" # session加盐
4. # request.form # 请求体
5. # request.args # 请求头
6. # request.method # 请求类型
7. @app.route("/login", methods=["GET", "POST"])
8. def login():
9. if request.method == "GET":
10. return render_template("login.html")
11. user = request.from.get("user")
12. pwd= request.from.get("pwd")
13. if user == "alex" and pwd == "666":
14. session["user"] = user
15. return redirect("/index")
16. return render_template("login.html", error="用户名或密码错误")
18. @app.route("/index")
19. def index():
20. user = session.get("user")
21. if not user:
22. return redirect("/login")
23. return render_template("index.html")
25. if __name__ == "__main__":
26. app.run()
复制代码

1.配置文件

settings
1. class Config(object):
2. DEBUG = False
3. TESTING = False
4. DATABASE_URL = "sqlite://:memory:"
6. class ProductionConfig(Config):
7. DATABASE_URL = "mysql://user@localhost/foo"
9. class DevelopmentConfig(Config):
10. DEBUT = True
12. class TestingConfig(Config):
13. TESTING = True
15. flask中的配置文件是一个flask.config.Config对象（继承字典）,默认配置为：
16. {
17. 'DEBUG': get_debug_flag(default=False), 是否开启Debug模式
18. 'TESTING': False, 是否开启测试模式
19. 'PROPAGATE_EXCEPTIONS': None,
20. 'PRESERVE_CONTEXT_ON_EXCEPTION': None,
21. 'SECRET_KEY': None,
22. 'PERMANENT_SESSION_LIFETIME': timedelta(days=31),
23. 'USE_X_SENDFILE': False,
24. 'LOGGER_NAME': None,
25. 'LOGGER_HANDLER_POLICY': 'always',
26. 'SERVER_NAME': None,
27. 'APPLICATION_ROOT': None,
28. 'SESSION_COOKIE_NAME': 'session',
29. 'SESSION_COOKIE_DOMAIN': None,
30. 'SESSION_COOKIE_PATH': None,
31. 'SESSION_COOKIE_HTTPONLY': True,
32. 'SESSION_COOKIE_SECURE': False,
33. 'SESSION_REFRESH_EACH_REQUEST': True,
34. 'MAX_CONTENT_LENGTH': None,
35. 'SEND_FILE_MAX_AGE_DEFAULT': timedelta(hours=12),
36. 'TRAP_BAD_REQUEST_ERRORS': False,
37. 'TRAP_HTTP_EXCEPTIONS': False,
38. 'EXPLAIN_TEMPLATE_LOADING': False,
39. 'PREFERRED_URL_SCHEME': 'http',
40. 'JSON_AS_ASCII': True,
41. 'JSON_SORT_KEYS': True,
42. 'JSONIFY_PRETTYPRINT_REGULAR': True,
43. 'JSONIFY_MIMETYPE': 'application/json',
44. 'TEMPLATES_AUTO_RELOAD': None,
45. }
复制代码
xx
1. # 如何通过代码找到指定的类
2. import importlib
3. path = "settings.Foo"
5. p, c = path.rsplit(".", maxsplit=1)
6. m = importlib.import_module(p)
7. cls = getattr(m, c)
8. # 找出类中的所有静态变量
9. for key in dir(cls):
10. if key.isupper():
11. print(key,getattr(cls, key))
复制代码
Flask中有一个配置文件
1. from flask import Flask, render_template, request, redirect, session
2. app = Flask(__name__)
4. print(app.config) # 这是flask提供的配置文件
5. # app.config.["DEBUG"] = True # 我们可以通过这样来修改配置，但是flask也提供了另一种方式
6. app.config.from_object("sttings.DevelopmentConfig") # 这种方式就是自己创建一个文件和类，你在类中写的所有配置都会被应用。
9. if __name__ == "__main__":
10. app.run()
复制代码

2.路由和视图

flask的路由系统
1. from flask import Flask, render_template, request, redirect, session， url_for
2. app = Flask(__name__)
3. # int 是值的类型，nid是名称：/login/1234
4. """
5. # 这是常用的类型，不加类型默认是字符串类型
6. DEFAULT_CONVERTERS = {
7. 'default': UnicodeConverter,
8. 'string': UnicodeConverter,
9. 'any': AnyConverter,
10. 'path': PathConverter,
11. 'int': IntegerConverter,
12. 'float': FloatConverter,
13. 'uuid': UUIDConverter,
14. }
15. """
16. @app.route("/login/<int:nid>", endpoint="n1") # endpoint就是设置名字，不写默认是函数名
17. def login():
18. url_for("n1", nid=122) # 反向查找url，nid是携带的值
20. if __name__ == "__main__":
21. app.run()
复制代码
路由的两种方式
1. @app.route("/xxx")
2. def index():
3. return "index"
4. # 其实到内部调用时就是使用了app.add_url_rule来实现的，为了方便所以在外面套了个装饰器。
5. def index():
6. return "index"
7. app.add_url_rule("/xxx", None, index)
9. # 注意事项：
10. # - 不用让endpoint重名
11. # - 如果重名函数也一定要相同。
复制代码
app.route和app.add_url_rule参数
1. @app.route和app.add_url_rule参数：
2. rule, URL规则
3. view_func, 视图函数名称
4. defaults=None, 默认值,当URL中无参数，函数需要参数时，使用defaults={'k':'v'}为函数提供参数
5. endpoint=None, 名称，用于反向生成URL，即： url_for('名称')
6. methods=None, 允许的请求方式，如：["GET","POST"]
8. strict_slashes=None, 对URL最后的 / 符号是否严格要求，
9. 如：
10. @app.route('/index',strict_slashes=False)，
11. 访问 http://www.xx.com/index/ 或 http://www.xx.com/index均可
12. @app.route('/index',strict_slashes=True)
13. 仅访问 http://www.xx.com/index
14. redirect_to=None, 重定向到指定地址
15. 如：
16. @app.route('/index/<int:nid>', redirect_to='/home/<nid>')
17. 或
18. def func(adapter, nid):
19. return "/home/888"
20. @app.route('/index/<int:nid>', redirect_to=func)
21. subdomain=None, 子域名访问
22. from flask import Flask, views, url_for
24. app = Flask(import_name=__name__)
25. app.config['SERVER_NAME'] = 'wupeiqi.com:5000'
27. @app.route("/", subdomain="admin")
28. def static_index():
29. """Flask supports static subdomains
30. This is available at static.your-domain.tld"""
31. return "static.your-domain.tld"
34. @app.route("/dynamic", subdomain="<username>")
35. def username_index(username):
36. """Dynamic subdomains are also supported
37. Try going to user1.your-domain.tld/dynamic"""
38. return username + ".your-domain.tld"
41. if __name__ == '__main__':
42. app.run()
复制代码
CBV
1. import functools
2. from flask import Flask,views
3. app = Flask(__name__)
6. def wrapper(func):
7. @functools.wraps(func)
8. def inner(*args,**kwargs):
9. return func(*args,**kwargs)
11. return inner
15. class UserView(views.MethodView):
16. methods = ['GET']
17. decorators = [wrapper,]
19. def get(self,*args,**kwargs):
20. return 'GET'
22. def post(self,*args,**kwargs):
23. return 'POST'
25. app.add_url_rule('/user',None,UserView.as_view('uuuu'))
27. if __name__ == '__main__':
28. app.run()
复制代码
自定义正则
1. from flask import Flask,url_for
3. app = Flask(__name__)
5. # 步骤一：定制类
6. from werkzeug.routing import BaseConverter
7. class RegexConverter(BaseConverter):
8. """
9. 自定义URL匹配正则表达式
10. """
12. def __init__(self, map, regex):
13. super(RegexConverter, self).__init__(map)
14. self.regex = regex
16. def to_python(self, value):
17. """
18. 路由匹配时，匹配成功后传递给视图函数中参数的值
19. :param value:
20. :return:
21. """
22. return int(value)
24. def to_url(self, value):
25. """
26. 使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数
27. :param value:
28. :return:
29. """
30. val = super(RegexConverter, self).to_url(value)
31. return val
33. # 步骤二：添加到转换器
34. app.url_map.converters['reg'] = RegexConverter
36. """
37. 1. 用户发送请求
38. 2. flask内部进行正则匹配
39. 3. 调用to_python(正则匹配的结果)方法
40. 4. to_python方法的返回值会交给视图函数的参数
41. """
43. # 步骤三：使用自定义正则
44. @app.route('/index/<reg("\d+"):nid>')
45. def index(nid):
46. print(nid,type(nid))
48. print(url_for('index',nid=987))
49. return "index"
51. if __name__ == '__main__':
52. app.run()
复制代码

3.请求和响应

请求
1. # request.method
2. # request.args
3. # request.form
4. # request.values
5. # request.cookies
6. # request.headers
7. # request.path
8. # request.full_path
9. # request.script_root
10. # request.url
11. # request.base_url
12. # request.url_root
13. # request.host_url
14. # request.host
15. # request.files
16. # obj = request.files['the_file_name']
17. # obj.save('/var/www/uploads/' + secure_filename(f.filename))
复制代码
响应
1. from flask import Flask, render_template, request, redirect, session， url_for, jsonify
3. # 响应相关信息
4. # return "字符串"
5. # return render_template('html模板路径',**{})
6. # return redirect('/index.html')
7. # return jsonify({"k1":"v1"})
9. # response = make_response(render_template('index.html'))
10. # response是flask.wrappers.Response类型
11. # response.delete_cookie('key')
12. # response.set_cookie('key', 'value')
13. # response.headers['X-Something'] = 'A value'
14. # return response
复制代码
实现登录后才可查看的效果
1. # 版本一：在每个函数中都加入判断
2. @app.route('/index')
3. def index():
4. if not session.get('user'):
5. return redirect(url_for('login'))
6. return render_template('index.html',stu_dic=STUDENT_DICT)
7. # 版本二：使用装饰器，但是需要用到functools的wraps，应为你写的装饰器肯定要先执行，不然无法被应用，但是你写的装饰器到最后被调用的函数是inner，但是我们需要自己的属性才可以实现。wraps的功能就是将inner赋予调用函数的所有属性和特征。
8. import functools
9. def auth(func):
10. @functools.wraps(func)
11. def inner(*args,**kwargs):
12. if not session.get('user'):
13. return redirect(url_for('login'))
14. ret = func(*args,**kwargs)
15. return ret
16. return inner
18. @app.route('/index')
19. @auth
20. def index():
21. return render_template('index.html',stu_dic=STUDENT_DICT)
23. # 应用场景：比较少的函数中需要额外添加功能。
25. # 版本三：before_request，在执行函数时会先执行before_request所定义的函数，返回None就会向后执行，如果返回数据就会直接将其返回到前端了。
26. @app.before_request
27. def xxxxxx():
28. if request.path == '/login':
29. return None
31. if session.get('user'):
32. return None
34. return redirect('/login')
复制代码

4.模板渲染

基本数据类型：可以执行python语法，如：dict.get() list['xx']
传入函数：django，自动执行\flask，不自动执行
全局定义函数
1. @app.template_global()
2. def sb(a1, a2):
3. # {{sb(1,9)}}
4. return a1 + a2
6. @app.template_filter()
7. def db(a1, a2, a3):
8. # {{ 1|db(2,3) }}
9. return a1 + a2 + a3
复制代码
模板继承
1. layout.html
2. <!DOCTYPE html>
3. <html lang="zh-CN">
4. <head>
5. <meta charset="UTF-8">
6. <title>Title</title>
7. <meta name="viewport" content="width=device-width, initial-scale=1">
8. </head>
9. <body>
10. <h1>模板</h1>
11. {% block content %}{% endblock %}
12. </body>
13. </html>
15. tpl.html
16. {% extends "layout.html"%}
19. {% block content %}
20. {{users.0}}
23. {% endblock %}
复制代码
include
1. {% include "form.html" %}
4. form.html
5. <form>
6. asdfasdf
7. asdfasdf
8. asdf
9. asdf
10. </form>
复制代码
宏
1. {% macro ccccc(name, type='text', value='') %}
2. <h1>宏</h1>
3. <input type="{{ type }}" name="{{ name }}" value="{{ value }}">
4. <input type="submit" value="提交">
5. {% endmacro %}
7. {{ ccccc('n1') }}
9. {{ ccccc('n2') }}
复制代码
flask默认时设置了xss的后端将html代码返回到前端时需要使用安全MarkUp("asdf")，前端也可以直接将其应用{{u|safe}}。

5.session

当请求刚到来：flask读取cookie中session对应的值：eyJrMiI6NDU2LCJ1c2VyIjoib2xkYm95，将该值解密并反序列化成字典，放入内存以便视图函数使用。
1. @app.route('/ses')
2. def ses():
3. session['k1'] = 123
4. session['k2'] = 456
5. del session['k1']
7. return "Session"
9. session['xxx'] = 123
10. session['xxx']
复制代码
当请求结束时，flask会读取内存中字典的值，进行序列化+加密，写入到用户cookie中。
threading.local【和flask无任何关系】
- 为每个线程创建一个独立的空间，使得线程对自己的空间中的数据进行操作（数据隔离）。
  1. import threading
  2. from threading import local
  3. import time
  5. obj = local()
  7. def task(i):
  8. obj.xxxxx = i
  9. time.sleep(2)
  10. print(obj.xxxxx,i)
  12. for i in range(10):
  13. t = threading.Thread(target=task,args=(i,))
  14. t.start()
  复制代码
- 根据字典自定义一个类似于threading.local功能？
  1. import time
  2. import threading
  4. DIC = {}
  6. def task(i):
  7. ident = threading.get_ident()
  8. if ident in DIC:
  9. DIC[ident]['xxxxx'] = i
  10. else:
  11. DIC[ident] = {'xxxxx':i }
  12. time.sleep(2)
  14. print(DIC[ident]['xxxxx'],i)
  16. for i in range(10):
  17. t = threading.Thread(target=task,args=(i,))
  18. t.start()
  复制代码
- 通过getattr/setattr 构造出来 threading.local的加强版（协程）
  1. import time
  2. import threading
  3. try:
  4. import greenlet
  5. get_ident = greenlet.getcurrent # 获取协程唯一标识
  6. except Exception as e:
  7. get_ident = threading.get_ident # 获取线程唯一标识
  9. class Local(object):
  10. DIC = {}
  12. def __getattr__(self, item): # obj.xxx时触发，item等于xxx
  13. ident = get_ident()
  14. if ident in self.DIC:
  15. return self.DIC[ident].get(item)
  16. return None
  18. def __setattr__(self, key, value): # obj.xxx = "123"时触发，key等于xxx,value等于123
  19. ident = get_ident()
  20. if ident in self.DIC:
  21. self.DIC[ident][key] = value
  22. else:
  23. self.DIC[ident] = {key:value}
  26. obj = Local()
  28. def task(i):
  29. obj.xxxxx = i
  30. time.sleep(2)
  31. print(obj.xxxxx,i)
  33. for i in range(10):
  34. t = threading.Thread(target=task,args=(i,))
  35. t.start()
  复制代码

5.1 flask-session

安装：pip3 install flask-session
使用
1. import redis
2. from flask import Flask,request,session
3. from flask.sessions import SecureCookieSessionInterface
4. from flask_session import Session
6. app = Flask(__name__)
8. # app.session_interface = SecureCookieSessionInterface()
9. # app.session_interface = RedisSessionInterface()
10. app.config['SESSION_TYPE'] = 'redis'
11. app.config['SESSION_REDIS'] = redis.Redis(host='140.143.227.206',port=6379,password='1234')
12. Session(app)
14. @app.route('/login')
15. def login():
16. session['user'] = 'alex'
17. return 'asdfasfd'
19. @app.route('/home')
20. def index():
21. print(session.get('user'))
23. return '...'
26. if __name__ == '__main__':
27. app.run()
复制代码

6.闪现和中间件

闪现，在session中存储一个数据，读取时通过pop将数据移除。
1. from flask import Flask,flash,get_flashed_messages
2. @app.route('/page1')
3. def page1():
4. flash('临时数据存储','error')
5. flash('sdfsdf234234','error')
6. flash('adasdfasdf','info')
7. return "Session"
9. @app.route('/page2')
10. def page2():
11. print(get_flashed_messages(category_filter=['error']))
12. return "Session"
复制代码
中间件
1. call方法什么时候出发？
2. 用户发起请求时，才执行。
3. 任务：在执行call方法之前，做一个操作，call方法执行之后做一个操作。
4. class Middleware(object):
5. def __init__(self,old):
6. self.old = old
8. def __call__(self, *args, **kwargs):
9. ret = self.old(*args, **kwargs)
10. return ret
12. if __name__ == '__main__':
13. app.wsgi_app = Middleware(app.wsgi_app)
14. app.run()
复制代码

7.蓝图

完整目录图
1. ProjectName
2. -ProjectName
3. --static
4. --templates
5. --views
6. ---account.py
7. from flask import Blueprint,render_template
8. # 创建蓝图对象
9. ac = Blueprint('ac',__name__)
10. # 只有调用该文件中的页面时才会触发
11. @ac.before_request
12. def x1():
13. print('app.before_request')
16. @ac.route('/login')
17. def login():
18. return render_template('login.html')
21. @ac.route('/logout')
22. def logout():
23. return 'Logout'
24. --__init__.py
25. from flask import Flask
26. from .views.account import ac
27. from .views.user import uc
29. def create_app():
31. app = Flask(__name__)
33. # @app.before_request
34. # def x1():
35. # print('app.before_request')
37. # 注册蓝图
38. app.register_blueprint(ac)
39. # 注册蓝图和编写url前缀
40. app.register_blueprint(uc,url_prefix='/api')
41. return app
44. -manage.py
45. from crm import create_app
47. app = create_app()
49. if __name__ == '__main__':
50. app.run()
复制代码

8.特殊装饰器

before_request and after_request
1. from flask import Flask
2. app = Flask(__name__)
5. @app.before_request
6. def x1():
7. print('before:x1')
8. return '滚'
10. @app.before_request
11. def xx1():
12. print('before:xx1')
15. @app.after_request
16. def x2(response):
17. print('after:x2')
18. return response
20. @app.after_request
21. def xx2(response):
22. print('after:xx2')
23. return response
26. @app.route('/index')
27. def index():
28. print('index')
29. return "Index"
32. @app.route('/order')
33. def order():
34. print('order')
35. return "order"
38. if __name__ == '__main__':
40. app.run()
42. # 执行结果是，before的返回会将其返回到页面，但是after还是会执行，并不像django返回后后面的都不会执行
43. before:x1
44. before:xx1
45. request
46. after:xx2
47. after:x2
复制代码
before_first_request，只有第一次时才会被触发
1. from flask import Flask
2. app = Flask(__name__)
4. @app.before_first_request
5. def x1():
6. print('123123')
9. @app.route('/index')
10. def index():
11. print('index')
12. return "Index"
15. @app.route('/order')
16. def order():
17. print('order')
18. return "order"
20. if __name__ == '__main__':
21. app.run()
复制代码
页面渲染的全局装饰器
1. @app.template_global()
2. def sb(a1, a2):
3. # {{sb(1,9)}}
4. return a1 + a2
6. @app.template_filter()
7. def db(a1, a2, a3):
8. # {{ 1|db(2,3) }}
9. return a1 + a2 + a3
复制代码
errorhandler，当报指定的错误时会触发。
1. @app.errorhandler(404)
2. def not_found(arg):
3. print(arg)
4. return "没找到"
复制代码

9.上下文管理

偏函数
1. import functools
3. def index(a1,a2):
4. return a1 + a2
6. # 原来的调用方式
7. # ret = index(1,23)
8. # print(ret)
10. # 偏函数，帮助开发者自动传递参数
11. new_func = functools.partial(index,666)
12. ret = new_func(1)
13. print(ret)
复制代码
父类方法的执行
1. """
2. class Base(object):
4. def func(self):
5. print('Base.func')
7. class Foo(Base):
9. def func(self):
10. # 方式一：根据mro的顺序执行方法
11. # super(Foo,self).func()
12. # 方式二：主动执行Base类的方法
13. # Base.func(self)
15. print('Foo.func')
18. obj = Foo()
19. obj.func()
20. """
21. ####################################
22. class Base(object):
24. def func(self):
25. super(Base, self).func()
26. print('Base.func')
28. class Bar(object):
29. def func(self):
30. print('Bar.func')
32. class Foo(Base,Bar):
33. pass
35. # 示例一
36. # obj = Foo()
37. # obj.func()
38. # print(Foo.__mro__)
40. # 示例二
41. # obj = Base()
42. # obj.func()
复制代码
栈
1. # by luffycity.com
3. class Stack(object):
5. def __init__(self):
6. self.data = []
8. def push(self,val):
9. self.data.append(val)
11. def pop(self):
12. return self.data.pop()
14. def top(self):
15. return self.data[-1]
17. _stack = Stack()
19. _stack.push('佳俊')
20. _stack.push('咸鱼')
22. print(_stack.pop())
23. print(_stack.pop())
复制代码
Local
1. """
2. {
3. 1232:{k:v}
4. }
7. """
8. try:
9. from greenlet import getcurrent as get_ident
10. except:
11. from threading import get_ident
13. """
14. class Local(object):
16. def __init__(self):
17. object.__setattr__(self,'storage',{})
19. def __setattr__(self, key, value):
20. ident = get_ident()
21. if ident not in self.storage:
22. self.storage[ident] = {key:value}
23. else:
24. self.storage[ident][key] = value
26. def __getattr__(self, item):
27. ident = get_ident()
28. if ident in self.storage:
29. return self.storage[ident].get(item)
30. """
33. class Local(object):
34. __slots__ = ('__storage__', '__ident_func__')
36. def __init__(self):
37. # __storage__ = {1231:{'stack':[]}}
38. object.__setattr__(self, '__storage__', {})
39. object.__setattr__(self, '__ident_func__', get_ident)
41. def __getattr__(self, name):
42. try:
43. return self.__storage__[self.__ident_func__()][name]
44. except KeyError:
45. raise AttributeError(name)
47. def __setattr__(self, name, value):
48. ident = self.__ident_func__()
49. storage = self.__storage__
50. try:
51. storage[ident][name] = value
52. except KeyError:
53. storage[ident] = {name: value}
55. def __delattr__(self, name):
56. try:
57. del self.__storage__[self.__ident_func__()][name]
58. except KeyError:
59. raise AttributeError(name)
63. obj = Local()
64. obj.stack = []
65. obj.stack.append('佳俊')
66. obj.stack.append('咸鱼')
67. print(obj.stack)
68. print(obj.stack.pop())
69. print(obj.stack)
复制代码
LocalStack
1. import functools
2. try:
3. from greenlet import getcurrent as get_ident
4. except:
5. from threading import get_ident
7. class Local(object):
8. __slots__ = ('__storage__', '__ident_func__')
10. def __init__(self):
11. # __storage__ = {1231:{'stack':[]}}
12. object.__setattr__(self, '__storage__', {})
13. object.__setattr__(self, '__ident_func__', get_ident)
15. def __getattr__(self, name):
16. try:
17. return self.__storage__[self.__ident_func__()][name]
18. except KeyError:
19. raise AttributeError(name)
21. def __setattr__(self, name, value):
22. # name=stack
23. # value=[]
24. ident = self.__ident_func__()
25. storage = self.__storage__
26. try:
27. storage[ident][name] = value
28. except KeyError:
29. storage[ident] = {name: value}
31. def __delattr__(self, name):
32. try:
33. del self.__storage__[self.__ident_func__()][name]
34. except KeyError:
35. raise AttributeError(name)
38. """
39. __storage__ = {
40. 12312: {stack:[ctx(session/request) ,]}
41. }
43. """
45. # obj = Local()
46. # obj.stack = []
47. # obj.stack.append('佳俊')
48. # obj.stack.append('咸鱼')
49. # print(obj.stack)
50. # print(obj.stack.pop())
51. # print(obj.stack)
54. class LocalStack(object):
55. def __init__(self):
56. self._local = Local()
58. def push(self,value):
59. rv = getattr(self._local, 'stack', None) # self._local.stack =>local.getattr
60. if rv is None:
61. self._local.stack = rv = [] # self._local.stack =>local.setattr
62. rv.append(value) # self._local.stack.append(666)
63. return rv
66. def pop(self):
67. """Removes the topmost item from the stack, will return the
68. old value or `None` if the stack was already empty.
69. """
70. stack = getattr(self._local, 'stack', None)
71. if stack is None:
72. return None
73. elif len(stack) == 1:
74. return stack[-1]
75. else:
76. return stack.pop()
78. def top(self):
79. try:
80. return self._local.stack[-1]
81. except (AttributeError, IndexError):
82. return None
86. class RequestContext(object):
87. def __init__(self):
88. self.request = "xx"
89. self.session = 'oo'
93. _request_ctx_stack = LocalStack()
95. _request_ctx_stack.push(RequestContext())
98. def _lookup_req_object(arg):
100. ctx = _request_ctx_stack.top()
102. return getattr(ctx,arg) # ctx.request / ctx.session
104. request = functools.partial(_lookup_req_object,'request')
105. session = functools.partial(_lookup_req_object,'session')
108. print(request())
109. print(session())
复制代码
slots
1. class Foo(object):
2. __slots__ = ('name',)
3. def __init__(self):
4. self.name = 'alex'
5. # self.age = 18
8. obj = Foo()
9. print(obj.name)
10. # print(obj.age)
复制代码
请求上下文管理（ctx）：request，session
1. - 请求到来之后wsgi会触发__call__方法，由__call__方法再次调用wsgi_app方法
2. - 在wsgi_app方法中：
3. - 首先将请求相关+空session 封装到一个RequestContext对象中，即：ctx。
4. - 将ctx交给LocalStack对象，再由LocalStack将ctx添加到Local中，Local结构：
5. __storage__ = {
6. 1231:{stack:[ctx,] }
7. }
8. - 根据请求中的cookie中提取名称为sessionid对应的值，对cookie进行加密+反序列化，再次赋值给ctx中的session
10. -> 视图函数
12. - 把session中的数据再次写入到cookie中。
13. - 将ctx删除
14. - 结果返回给用户浏览器
15. - 断开socket连接
复制代码
LocalProxy
1. from flask import Flask,request,session
3. app = Flask(__name__)
6. @app.route('/index')
7. def index():
8. # 1. request是LocalProxy对象
9. # 2. 对象中有method、执行__getattr__
10. print(request.method)
11. # request['method']
12. # request + 1
14. # 1. session是LocalProxy对象
15. # 2. LocalProxy对象的__setitem__
16. session['x'] = 123
18. return "Index"
21. if __name__ == '__main__':
22. app.run()
23. # app.__call__
24. # app.wsgi_app
26. """
27. 第一阶段：请求到来
28. 将request和Session相关数据封装到ctx=RequestContext对象中。
29. 再通过LocalStack将ctx添加到Local中。
30. __storage__ = {
31. 1231:{'stack':[ctx(request,session)]}
32. }
33. 第二阶段：视图函数中获取request或session
34. 方式一：直接找LocalStack获取
35. from flask.globals import _request_ctx_stack
36. print(_request_ctx_stack.top.request.method)
38. 方式二：通过代理LocalProxy(小东北)获取
39. from flask import Flask,request
40. print(request.method)
42. """
复制代码
g只存在在一次请求中当你return时g就会被清除。

10.文件的上传和解压

view.py
1. from flask import Blueprint, render_template, Flask, request, redirect,session
2. import os
3. import uuid
4. from ..utils import helper
7. ind = Blueprint('ind', __name__)
8. ind.config["MAX_CONTENT_LENGTH"] = = 1024 * 1024 * 7
9. @ind.before_request
10. def process_request():
11. if not session.get("user_info"):
12. return redirect("/login")
13. return None
16. @ind.route('/home')
17. def home():
18. return render_template('home.html')
20. @ind.route('/user_list')
21. def user_list():
22. # import pymysql
23. # conn = pymysql.Connect(host='127.0.0.1', user='root', password='123456', database='s9day118', charset='utf8')
24. # cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
25. # cursor.execute("SELECT id,user,nickname FROM userinfo")
26. # data_list = cursor.fetchall()
27. # cursor.close()
28. # conn.close()
29. data_list = helper.fetch_all("SELECT id,user,nickname FROM userinfo",[])
31. return render_template('user_list.html',data_list=data_list)
33. @ind.route('/detail/<int:nid>')
34. def detail(nid):
35. # import pymysql
36. # conn = Config.POOL.connection()
37. # cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
38. # cursor.execute("SELECT id,line,ctime FROM record where user_id=%s",(nid,))
39. # record_list = cursor.fetchall()
40. # cursor.close()
41. # conn.close()
42. record_list = helper.fetch_all("SELECT id,line,ctime FROM record where user_id=%s",(nid,))
44. return render_template('detail.html',record_list=record_list)
46. @ind.route('/upload',methods=['GET','POST'])
47. def upload():
48. if request.method == "GET":
49. return render_template('upload.html')
50. from werkzeug.datastructures import FileStorage
51. file_obj = request.files.get('code')
53. # 1. 检查上传文件后缀名
54. name_ext = file_obj.filename.rsplit('.',maxsplit=1)
55. if len(name_ext) != 2:
56. return "请上传zip压缩文件"
57. if name_ext[1] != 'zip':
58. return "请上传zip压缩文件"
60. """
61. # 2. 接收用户上传文件,并写入到服务器本地.
62. file_path = os.path.join("files",file_obj.filename)
63. # 从file_obj.stream中读取内容，写入到文件
64. file_obj.save(file_path)
66. # 3. 解压zip文件
67. import shutil
68. # 通过open打开压缩文件，读取内容再进行解压。
69. shutil._unpack_zipfile(file_path,'xsadfasdfasdf')
70. """
72. # 2+3, 接收用户上传文件，并解压到指定目录
73. import shutil
74. target_path = os.path.join('files',str(uuid.uuid4()))
75. shutil._unpack_zipfile(file_obj.stream,target_path)
77. # 4. 遍历某目录下的所有文件
78. # for item in os.listdir(target_path):
79. # print(item)
80. total_num = 0
81. for base_path,folder_list,file_list in os.walk(target_path):
82. for file_name in file_list:
83. file_path = os.path.join(base_path,file_name)
84. file_ext = file_path.rsplit('.',maxsplit=1)
85. if len(file_ext) != 2:
86. continue
87. if file_ext[1] != 'py':
88. continue
89. file_num = 0
90. with open(file_path,'rb') as f:
91. for line in f:
92. line = line.strip()
93. if not line:
94. continue
95. if line.startswith(b'#'):
96. continue
97. file_num += 1
98. total_num += file_num
100. # 获取当前时间
101. import datetime
102. ctime = datetime.date.today()
103. print(total_num,ctime,session['user_info']['id'])
105. # import pymysql
106. # conn = pymysql.Connect(host='127.0.0.1', user='root', password='123456', database='s9day118', charset='utf8')
107. # cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
108. # cursor.execute("select id from record where ctime=%s and user_id=%s",(ctime,session['user_info']['id']))
109. # data = cursor.fetchone()
110. # cursor.close()
111. # conn.close()
112. data = helper.fetch_one("select id from record where ctime=%s and user_id=%s",(ctime,session['user_info']['id']))
113. if data:
114. return "今天已经上传"
117. # import pymysql
118. # conn = pymysql.Connect(host='127.0.0.1', user='root', password='123456', database='s9day118', charset='utf8')
119. # cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
120. # cursor.execute("insert into record(line,ctime,user_id)values(%s,%s,%s)",(total_num,ctime,session['user_info']['id']))
121. # conn.commit()
122. # cursor.close()
123. # conn.close()
124. helper.insert("insert into record(line,ctime,user_id)values(%s,%s,%s)",(total_num,ctime,session['user_info']['id']))
126. return "上传成功"
复制代码
html.py
1. {% extends "layout.html"%}
3. {% block content %}
5. <h1>代码上传</h1>
7. <form method="post" enctype="multipart/form-data">
8. <input type="file" name="code">
9. <input type="submit" value="上传">
10. </form>
13. {% endblock %}
复制代码

11.Database Join 池

安装pip3 install DBUtils
创建连接池可以共享出去，谁要用就来获取一个连接，这样可以管理连接的数量，提高链接的利用率。
1. from DBUtils.PooledDB import PooledDB, SharedDBConnection
2. import pymysql
4. POOL = PooledDB(
5. creator=pymysql, # 使用链接数据库的模块
6. maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数
7. mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
8. maxcached=5, # 链接池中最多闲置的链接，0和None不限制
9. maxshared=3,
10. # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
11. blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
12. maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制
13. setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
14. ping=0,
15. # 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
16. host='127.0.0.1',
17. port=3306,
18. user='root',
19. password='123456',
20. database='s9day118',
21. charset='utf8'
22. )
23. # 获取连接
24. conn = POOL.connection()
25. cursor = conn.cursor(pymysql.cursors.DictCursor)
26. cursor.execute("select * from tb1")
27. result = cursor.fetchall()
28. cursor.close()
29. # 归还连接
30. conn.close()
复制代码

12.wtforms

WTForms是一个支持多个web框架的form组件，主要用于对用户请求数据进行验证。
安装pip3 install wtforms

12.1 用户登录注册示例

用户登录
- 当用户登录时候，需要对用户提交的用户名和密码进行多种格式校验。如：
- 用户不能为空；用户长度必须大于6；
- 密码不能为空；密码长度必须大于12；密码必须包含字母、数字、特殊字符等（自定义正则）；
  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. from flask import Flask, render_template, request, redirect
  4. from wtforms import Form
  5. from wtforms.fields import core
  6. from wtforms.fields import html5
  7. from wtforms.fields import simple
  8. from wtforms import validators
  9. from wtforms import widgets
  11. app = Flask(__name__, template_folder='templates')
  12. app.debug = True
  15. class LoginForm(Form):
  16. name = simple.StringField(
  17. label='用户名',
  18. validators=[
  19. validators.DataRequired(message='用户名不能为空.'),
  20. validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d')
  21. ],
  22. widget=widgets.TextInput(),
  23. render_kw={'class': 'form-control'}
  25. )
  26. pwd = simple.PasswordField(
  27. label='密码',
  28. validators=[
  29. validators.DataRequired(message='密码不能为空.'),
  30. validators.Length(min=8, message='用户名长度必须大于%(min)d'),
  31. validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[$@$!%*?&])[A-Za-z\d$@$!%*?&]{8,}",
  32. message='密码至少8个字符，至少1个大写字母，1个小写字母，1个数字和1个特殊字符')
  34. ],
  35. widget=widgets.PasswordInput(),
  36. render_kw={'class': 'form-control'}
  37. )
  41. @app.route('/login', methods=['GET', 'POST'])
  42. def login():
  43. if request.method == 'GET':
  44. form = LoginForm()
  45. return render_template('login.html', form=form)
  46. else:
  47. form = LoginForm(formdata=request.form)
  48. if form.validate():
  49. print('用户提交数据通过格式验证，提交的值为：', form.data)
  50. else:
  51. print(form.errors)
  52. return render_template('login.html', form=form)
  54. if __name__ == '__main__':
  55. app.run()
  56. #######################################################
  58. <!DOCTYPE html>
  59. <html lang="en">
  60. <head>
  61. <meta charset="UTF-8">
  62. <title>Title</title>
  63. </head>
  64. <body>
  65. <h1>登录</h1>
  66. <form method="post">
  68. <p>{{form.name.label}} {{form.name}} {{form.name.errors[0] }}</p>
  71. <p>{{form.pwd.label}} {{form.pwd}} {{form.pwd.errors[0] }}</p>
  72. <input type="submit" value="提交">
  73. </form>
  74. </body>
  75. </html>
  复制代码
用户注册
- 注册页面需要让用户输入：用户名、密码、密码重复、性别、爱好等。
  1. from flask import Flask, render_template, request, redirect
  2. from wtforms import Form
  3. from wtforms.fields import core
  4. from wtforms.fields import html5
  5. from wtforms.fields import simple
  6. from wtforms import validators
  7. from wtforms import widgets
  9. app = Flask(__name__, template_folder='templates')
  10. app.debug = True
  14. class RegisterForm(Form):
  15. name = simple.StringField(
  16. label='用户名',
  17. validators=[
  18. validators.DataRequired()
  19. ],
  20. widget=widgets.TextInput(),
  21. render_kw={'class': 'form-control'},
  22. default='alex'
  23. )
  25. pwd = simple.PasswordField(
  26. label='密码',
  27. validators=[
  28. validators.DataRequired(message='密码不能为空.')
  29. ],
  30. widget=widgets.PasswordInput(),
  31. render_kw={'class': 'form-control'}
  32. )
  34. pwd_confirm = simple.PasswordField(
  35. label='重复密码',
  36. validators=[
  37. validators.DataRequired(message='重复密码不能为空.'),
  38. validators.EqualTo('pwd', message="两次密码输入不一致")
  39. ],
  40. widget=widgets.PasswordInput(),
  41. render_kw={'class': 'form-control'}
  42. )
  44. email = html5.EmailField(
  45. label='邮箱',
  46. validators=[
  47. validators.DataRequired(message='邮箱不能为空.'),
  48. validators.Email(message='邮箱格式错误')
  49. ],
  50. widget=widgets.TextInput(input_type='email'),
  51. render_kw={'class': 'form-control'}
  52. )
  54. gender = core.RadioField(
  55. label='性别',
  56. choices=(
  57. (1, '男'),
  58. (2, '女'),
  59. ),
  60. coerce=int
  61. )
  62. city = core.SelectField(
  63. label='城市',
  64. choices=(
  65. ('bj', '北京'),
  66. ('sh', '上海'),
  67. )
  68. )
  70. hobby = core.SelectMultipleField(
  71. label='爱好',
  72. choices=(
  73. (1, '篮球'),
  74. (2, '足球'),
  75. ),
  76. coerce=int
  77. )
  79. favor = core.SelectMultipleField(
  80. label='喜好',
  81. choices=(
  82. (1, '篮球'),
  83. (2, '足球'),
  84. ),
  85. widget=widgets.ListWidget(prefix_label=False),
  86. option_widget=widgets.CheckboxInput(),
  87. coerce=int,
  88. default=[1, 2]
  89. )
  91. def __init__(self, *args, **kwargs):
  92. super(RegisterForm, self).__init__(*args, **kwargs)
  93. self.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球'))
  95. def validate_pwd_confirm(self, field):
  96. """
  97. 自定义pwd_confirm字段规则，例：与pwd字段是否一致
  98. :param field:
  99. :return:
  100. """
  101. # 最开始初始化时，self.data中已经有所有的值
  103. if field.data != self.data['pwd']:
  104. # raise validators.ValidationError("密码不一致") # 继续后续验证
  105. raise validators.StopValidation("密码不一致") # 不再继续后续验证
  108. @app.route('/register', methods=['GET', 'POST'])
  109. def register():
  110. if request.method == 'GET':
  111. form = RegisterForm(data={'gender': 1})
  112. return render_template('register.html', form=form)
  113. else:
  114. form = RegisterForm(formdata=request.form)
  115. if form.validate():
  116. print('用户提交数据通过格式验证，提交的值为：', form.data)
  117. else:
  118. print(form.errors)
  119. return render_template('register.html', form=form)
  123. if __name__ == '__main__':
  124. app.run()
  127. ############################################################
  129. <!DOCTYPE html>
  130. <html lang="en">
  131. <head>
  132. <meta charset="UTF-8">
  133. <title>Title</title>
  134. </head>
  135. <body>
  136. <h1>用户注册</h1>
  137. <form method="post" novalidate >
  138. {% for item in form %}
  139. <p>{{item.label}}: {{item}} {{item.errors[0] }}</p>
  140. {% endfor %}
  141. <input type="submit" value="提交">
  142. </form>
  143. </body>
  144. </html>
  复制代码
动态获取数据库数据
1. from flask import Flask,request,render_template,session,current_app,g,redirect
2. from wtforms import Form
3. from wtforms.fields import simple
4. from wtforms.fields import html5
5. from wtforms.fields import core
7. from wtforms import widgets
8. from wtforms import validators
10. app = Flask(__name__)
13. class LoginForm(Form):
14. name = simple.StringField(
15. validators=[
16. validators.DataRequired(message='用户名不能为空.'),
17. # validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d')
18. ],
19. widget=widgets.TextInput(),
20. render_kw={'placeholder':'请输入用户名'}
21. )
22. pwd = simple.PasswordField(
23. validators=[
24. validators.DataRequired(message='密码不能为空.'),
25. # validators.Length(min=8, message='用户名长度必须大于%(min)d'),
26. # validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[$@$!%*?&])[A-Za-z\d$@$!%*?&]{8,}",
27. # message='密码至少8个字符，至少1个大写字母，1个小写字母，1个数字和1个特殊字符')
29. ],
30. render_kw={'placeholder':'请输入密码'}
31. )
34. @app.route('/login',methods=['GET','POST'])
35. def login():
36. if request.method == "GET":
37. form = LoginForm()
38. # print(form.name,type(form.name)) # form.name是StringField()对象, StringField().__str__
39. # print(form.pwd,type(form.pwd)) # form.pwd是PasswordField()对象,PasswordField().__str__
40. return render_template('login.html',form=form)
42. form = LoginForm(formdata=request.form)
43. if form.validate():
44. print(form.data)
45. return redirect('https://www.luffycity.com/home')
46. else:
47. # print(form.errors)
48. return render_template('login.html', form=form)
51. import helper
52. class UserForm(Form):
53. city = core.SelectField(
54. label='城市',
55. choices=(),
56. coerce=int # 从数据库获取的数据返回后是字符串类型的，把他写上后会先将其转为int后再返回到前端
57. )
58. name = simple.StringField(label='姓名')
59. # 当你定义成静态字段是，只有启动时才会加载一次，但是从数据库需要实时拿到数据，下面是解决方法。你应该可以看懂。
60. def __init__(self,*args,**kwargs):
61. super(UserForm,self).__init__(*args,**kwargs)
63. self.city.choices=helper.fetch_all('select id,name from tb1',[],type=None)
66. @app.route('/user')
67. def user():
68. if request.method == "GET":
69. #form = UserForm(data={'name':'alex','city':3})
70. form = UserForm()
71. return render_template('user.html',form=form)
74. if __name__ == '__main__':
75. app.run()
复制代码
meta
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. from flask import Flask, render_template, request, redirect, session
4. from wtforms import Form
5. from wtforms.csrf.core import CSRF
6. from wtforms.fields import core
7. from wtforms.fields import html5
8. from wtforms.fields import simple
9. from wtforms import validators
10. from wtforms import widgets
11. from hashlib import md5
13. app = Flask(__name__, template_folder='templates')
14. app.debug = True
17. class MyCSRF(CSRF):
18. """
19. Generate a CSRF token based on the user's IP. I am probably not very
20. secure, so don't use me.
21. """
23. def setup_form(self, form):
24. self.csrf_context = form.meta.csrf_context()
25. self.csrf_secret = form.meta.csrf_secret
26. return super(MyCSRF, self).setup_form(form)
28. def generate_csrf_token(self, csrf_token):
29. gid = self.csrf_secret + self.csrf_context
30. token = md5(gid.encode('utf-8')).hexdigest()
31. return token
33. def validate_csrf_token(self, form, field):
34. print(field.data, field.current_token)
35. if field.data != field.current_token:
36. raise ValueError('Invalid CSRF')
39. class TestForm(Form):
40. name = html5.EmailField(label='用户名')
41. pwd = simple.StringField(label='密码')
43. class Meta:
44. # -- CSRF
45. # 是否自动生成CSRF标签
46. csrf = True
47. # 生成CSRF标签name
48. csrf_field_name = 'csrf_token'
50. # 自动生成标签的值，加密用的csrf_secret
51. csrf_secret = 'xxxxxx'
52. # 自动生成标签的值，加密用的csrf_context
53. csrf_context = lambda x: request.url
54. # 生成和比较csrf标签
55. csrf_class = MyCSRF
57. # -- i18n
58. # 是否支持本地化
59. # locales = False
60. locales = ('zh', 'en')
61. # 是否对本地化进行缓存
62. cache_translations = True
63. # 保存本地化缓存信息的字段
64. translations_cache = {}
67. @app.route('/index/', methods=['GET', 'POST'])
68. def index():
69. if request.method == 'GET':
70. form = TestForm()
71. else:
72. form = TestForm(formdata=request.form)
73. if form.validate():
74. print(form)
75. return render_template('index.html', form=form)
78. if __name__ == '__main__':
79. app.run()
复制代码

12.2 其他

metaclass
1. # new方法的返回值决定对象到底是什么？
2. class Bar(object):
3. pass
5. class Foo(object):
7. def __new__(cls, *args, **kwargs):
8. # return super(Foo,cls).__new__(cls,*args, **kwargs)
9. return Bar()
10. obj = Foo()
11. print(obj)
12. # 当Foo()时先执行__new__方法，返回一个Bar，obj就等于Bar对象
16. ##########################################################################
21. class MyType(type):
22. def __init__(self, *args, **kwargs):
23. print('MyType创建类',self)
24. super(MyType, self).__init__(*args, **kwargs)
26. def __call__(self, *args, **kwargs):
27. obj = super(MyType, self).__call__(*args, **kwargs)
28. print('类创建对象', self, obj)
29. return obj
32. class Foo(object,metaclass=MyType):
33. user = 'wupeiqi'
34. age = 18
36. obj = Foo()
37. # 在创建Foo对象前先执行mytype的__call__方法，__call__方法后才会执行Foo。不写metaclass默认是由type类来做的。
38. # 是先执行父类的__call__后再执行自己的__init__方法。
40. """
41. 创建类时，先执行type的__init__。
42. 类的实例化时，执行type的__call__，__call__方法的的返回值就是实例化的对象。
43. __call__内部调用：
44. 类.__new__，创建对象
45. 类.__init__，对象的初始化
46. """
复制代码
实例化流程分析
1. # 源码流程
2. 1. 执行type的 __call__ 方法，读取字段到静态字段 cls._unbound_fields 中； meta类读取到cls._wtforms_meta中
3. 2. 执行构造方法
5. a. 循环cls._unbound_fields中的字段，并执行字段的bind方法，然后将返回值添加到 self._fields[name] 中。
6. 即：
7. _fields = {
8. name: wtforms.fields.core.StringField(),
9. }
11. PS：由于字段中的__new__方法，实例化时：name = simple.StringField(label='用户名')，创建的是UnboundField(cls, *args, **kwargs)，当执行完bind之后，才变成执行 wtforms.fields.core.StringField()
13. b. 循环_fields，为对象设置属性
14. for name, field in iteritems(self._fields):
15. # Set all the fields to attributes so that they obscure the class
16. # attributes with the same names.
17. setattr(self, name, field)
18. c. 执行process，为字段设置默认值：self.process(formdata, obj, data=data, **kwargs)
19. 优先级：obj,data,formdata;
21. 再循环执行每个字段的process方法，为每个字段设置值：
22. for name, field, in iteritems(self._fields):
23. if obj is not None and hasattr(obj, name):
24. field.process(formdata, getattr(obj, name))
25. elif name in kwargs:
26. field.process(formdata, kwargs[name])
27. else:
28. field.process(formdata)
30. 执行每个字段的process方法，为字段的data和字段的raw_data赋值
31. def process(self, formdata, data=unset_value):
32. self.process_errors = []
33. if data is unset_value:
34. try:
35. data = self.default()
36. except TypeError:
37. data = self.default
39. self.object_data = data
41. try:
42. self.process_data(data)
43. except ValueError as e:
44. self.process_errors.append(e.args[0])
46. if formdata:
47. try:
48. if self.name in formdata:
49. self.raw_data = formdata.getlist(self.name)
50. else:
51. self.raw_data = []
52. self.process_formdata(self.raw_data)
53. except ValueError as e:
54. self.process_errors.append(e.args[0])
56. try:
57. for filter in self.filters:
58. self.data = filter(self.data)
59. except ValueError as e:
60. self.process_errors.append(e.args[0])
62. d. 页面上执行print(form.name) 时，打印标签
64. 因为执行了：
65. 字段的 __str__ 方法
66. 字符的 __call__ 方法
67. self.meta.render_field(self, kwargs)
68. def render_field(self, field, render_kw):
69. other_kw = getattr(field, 'render_kw', None)
70. if other_kw is not None:
71. render_kw = dict(other_kw, **render_kw)
72. return field.widget(field, **render_kw)
73. 执行字段的插件对象的 __call__ 方法，返回标签字符串
复制代码
验证流程分析
1. a. 执行form的validate方法，获取钩子方法
2. def validate(self):
3. extra = {}
4. for name in self._fields:
5. inline = getattr(self.__class__, 'validate_%s' % name, None)
6. if inline is not None:
7. extra[name] = [inline]
9. return super(Form, self).validate(extra)
10. b. 循环每一个字段，执行字段的 validate 方法进行校验(参数传递了钩子函数)
11. def validate(self, extra_validators=None):
12. self._errors = None
13. success = True
14. for name, field in iteritems(self._fields):
15. if extra_validators is not None and name in extra_validators:
16. extra = extra_validators[name]
17. else:
18. extra = tuple()
19. if not field.validate(self, extra):
20. success = False
21. return success
22. c. 每个字段进行验证时候
23. 字段的pre_validate 【预留的扩展】
24. 字段的_run_validation_chain，对正则和字段的钩子函数进行校验
25. 字段的post_validate【预留的扩展】
复制代码

13.SQLAlchemy

MySQLdb，只支持python2.0的，而pymysql是支持python2.0和3.0的。
1. import MySQLdb
3. conn = MySQLdb.connect(host='127.0.0.1',user='root',passwd='1234',db='mydb')
5. cur = conn.cursor()
7. reCount = cur.execute('insert into UserInfo(Name,Address) values(%s,%s)',('alex','usa'))
8. # reCount = cur.execute('insert into UserInfo(Name,Address) values(%(id)s, %(name)s)',{'id':12345,'name':'wupeiqi'})
10. conn.commit()
12. cur.close()
13. conn.close()
15. print reCount
复制代码
SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
安装pip3 install sqlalchemy
组成部分：
- Engine，框架的引擎
- Connection Pooling ，数据库连接池
- Dialect，选择连接数据库的DB API种类
- Schema/Types，架构和类型
- SQL Exprression Language，SQL表达式语言
SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：
1. MySQL-Python
2. mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
4. pymysql
5. mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
7. MySQL-Connector
8. mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
10. cx_Oracle
11. oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
13. 更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html
复制代码
执行原生SQL语句
1. import time
2. import threading
3. import sqlalchemy
4. from sqlalchemy import create_engine
5. from sqlalchemy.engine.base import Engine
7. engine = create_engine(
8. "mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",
9. max_overflow=0, # 超过连接池大小外最多创建的连接
10. pool_size=5, # 连接池大小
11. pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
12. pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
13. )
16. def task(arg):
17. conn = engine.raw_connection()
18. cursor = conn.cursor()
19. cursor.execute(
20. "select * from t1"
21. )
22. result = cursor.fetchall()
23. cursor.close()
24. conn.close()
27. for i in range(20):
28. t = threading.Thread(target=task, args=(i,))
29. t.start()
32. ########################################################################################
34. #!/usr/bin/env python
35. # -*- coding:utf-8 -*-
36. import time
37. import threading
38. import sqlalchemy
39. from sqlalchemy import create_engine
40. from sqlalchemy.engine.base import Engine
42. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)
45. def task(arg):
46. conn = engine.contextual_connect()
47. with conn:
48. cur = conn.execute(
49. "select * from t1"
50. )
51. result = cur.fetchall()
52. print(result)
55. for i in range(20):
56. t = threading.Thread(target=task, args=(i,))
57. t.start()
60. ########################################################################################
63. #!/usr/bin/env python
64. # -*- coding:utf-8 -*-
65. import time
66. import threading
67. import sqlalchemy
68. from sqlalchemy import create_engine
69. from sqlalchemy.engine.base import Engine
70. from sqlalchemy.engine.result import ResultProxy
71. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)
74. def task(arg):
75. cur = engine.execute("select * from t1")
76. result = cur.fetchall()
77. cur.close()
78. print(result)
81. for i in range(20):
82. t = threading.Thread(target=task, args=(i,))
83. t.start()
86. # 注意: 查看连接 show status like 'Threads%';
复制代码

13.1 ORM

创建单表
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import datetime
4. from sqlalchemy import create_engine
5. from sqlalchemy.ext.declarative import declarative_base
6. from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
8. Base = declarative_base()
11. class Users(Base):
12. __tablename__ = 'users'
14. id = Column(Integer, primary_key=True)
15. name = Column(String(32), index=True, nullable=False)
16. # email = Column(String(32), unique=True)
17. # ctime = Column(DateTime, default=datetime.datetime.now)
18. # extra = Column(Text, nullable=True)
20. __table_args__ = (
21. # UniqueConstraint('id', 'name', name='uix_id_name'),
22. # Index('ix_id_name', 'name', 'email'),
23. )
26. def init_db():
27. """
28. 根据类创建数据库表
29. :return:
30. """
31. engine = create_engine(
32. "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
33. max_overflow=0, # 超过连接池大小外最多创建的连接
34. pool_size=5, # 连接池大小
35. pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
36. pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
37. )
39. Base.metadata.create_all(engine)
42. def drop_db():
43. """
44. 根据类删除数据库表
45. :return:
46. """
47. engine = create_engine(
48. "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
49. max_overflow=0, # 超过连接池大小外最多创建的连接
50. pool_size=5, # 连接池大小
51. pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
52. pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
53. )
55. Base.metadata.drop_all(engine)
58. if __name__ == '__main__':
59. drop_db()
60. init_db()
复制代码
创建多个表并包含Fk、M2M关系
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import datetime
4. from sqlalchemy import create_engine
5. from sqlalchemy.ext.declarative import declarative_base
6. from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
7. from sqlalchemy.orm import relationship
9. Base = declarative_base()
12. # ##################### 单表示例 #########################
13. class Users(Base):
14. __tablename__ = 'users'
16. id = Column(Integer, primary_key=True)
17. name = Column(String(32), index=True)
18. age = Column(Integer, default=18)
19. email = Column(String(32), unique=True)
20. ctime = Column(DateTime, default=datetime.datetime.now)
21. extra = Column(Text, nullable=True)
23. __table_args__ = (
24. # UniqueConstraint('id', 'name', name='uix_id_name'),
25. # Index('ix_id_name', 'name', 'extra'),
26. )
29. class Hosts(Base):
30. __tablename__ = 'hosts'
32. id = Column(Integer, primary_key=True)
33. name = Column(String(32), index=True)
34. ctime = Column(DateTime, default=datetime.datetime.now)
37. # ##################### 一对多示例 #########################
38. class Hobby(Base):
39. __tablename__ = 'hobby'
40. id = Column(Integer, primary_key=True)
41. caption = Column(String(50), default='篮球')
44. class Person(Base):
45. __tablename__ = 'person'
46. nid = Column(Integer, primary_key=True)
47. name = Column(String(32), index=True, nullable=True)
48. hobby_id = Column(Integer, ForeignKey("hobby.id"))
50. # 与生成表结构无关，仅用于查询方便
51. hobby = relationship("Hobby", backref='pers')
54. # ##################### 多对多示例 #########################
56. class Server2Group(Base):
57. __tablename__ = 'server2group'
58. id = Column(Integer, primary_key=True, autoincrement=True)
59. server_id = Column(Integer, ForeignKey('server.id'))
60. group_id = Column(Integer, ForeignKey('group.id'))
63. class Group(Base):
64. __tablename__ = 'group'
65. id = Column(Integer, primary_key=True)
66. name = Column(String(64), unique=True, nullable=False)
68. # 与生成表结构无关，仅用于查询方便
69. servers = relationship('Server', secondary='server2group', backref='groups')
72. class Server(Base):
73. __tablename__ = 'server'
75. id = Column(Integer, primary_key=True, autoincrement=True)
76. hostname = Column(String(64), unique=True, nullable=False)
79. def init_db():
80. """
81. 根据类创建数据库表
82. :return:
83. """
84. engine = create_engine(
85. "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
86. max_overflow=0, # 超过连接池大小外最多创建的连接
87. pool_size=5, # 连接池大小
88. pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
89. pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
90. )
92. Base.metadata.create_all(engine)
95. def drop_db():
96. """
97. 根据类删除数据库表
98. :return:
99. """
100. engine = create_engine(
101. "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
102. max_overflow=0, # 超过连接池大小外最多创建的连接
103. pool_size=5, # 连接池大小
104. pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
105. pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
106. )
108. Base.metadata.drop_all(engine)
111. if __name__ == '__main__':
112. drop_db()
113. init_db()
复制代码
操作数据库表
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. from sqlalchemy.orm import sessionmaker
4. from sqlalchemy import create_engine
5. from models import Users
7. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
8. Session = sessionmaker(bind=engine)
10. # 每次执行数据库操作时，都需要创建一个session
11. session = Session()
13. # ############# 执行ORM操作 #############
14. obj1 = Users(name="alex1")
15. session.add(obj1)
17. # 提交事务
18. session.commit()
19. # 关闭session
20. session.close()
复制代码
基于scoped_session实现线程安全
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. from sqlalchemy.orm import sessionmaker
4. from sqlalchemy import create_engine
5. from sqlalchemy.orm import scoped_session
6. from models import Users
8. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
9. Session = sessionmaker(bind=engine)
11. """
12. # 线程安全，基于本地线程实现每个线程用同一个session
13. # 特殊的：scoped_session中有原来方法的Session中的一下方法：
15. public_methods = (
16. '__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',
17. 'close', 'commit', 'connection', 'delete', 'execute', 'expire',
18. 'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind',
19. 'is_modified', 'bulk_save_objects', 'bulk_insert_mappings',
20. 'bulk_update_mappings',
21. 'merge', 'query', 'refresh', 'rollback',
22. 'scalar'
23. )
24. """
25. session = scoped_session(Session)
28. # ############# 执行ORM操作 #############
29. obj1 = Users(name="alex1")
30. session.add(obj1)
34. # 提交事务
35. session.commit()
36. # 关闭session
37. session.close()
复制代码
多线程执行示例
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import time
4. import threading
6. from sqlalchemy.ext.declarative import declarative_base
7. from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
8. from sqlalchemy.orm import sessionmaker, relationship
9. from sqlalchemy import create_engine
10. from db import Users
12. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
13. Session = sessionmaker(bind=engine)
16. def task(arg):
17. session = Session()
19. obj1 = Users(name="alex1")
20. session.add(obj1)
22. session.commit()
25. for i in range(10):
26. t = threading.Thread(target=task, args=(i,))
27. t.start()
复制代码
基本增删改查示例
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import time
4. import threading
6. from sqlalchemy.ext.declarative import declarative_base
7. from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
8. from sqlalchemy.orm import sessionmaker, relationship
9. from sqlalchemy import create_engine
10. from sqlalchemy.sql import text
12. from db import Users, Hosts
14. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
15. Session = sessionmaker(bind=engine)
17. session = Session()
19. # ################ 添加 ################
20. """
21. obj1 = Users(name="wupeiqi")
22. session.add(obj1)
24. session.add_all([
25. Users(name="wupeiqi"),
26. Users(name="alex"),
27. Hosts(name="c1.com"),
28. ])
29. session.commit()
30. """
32. # ################ 删除 ################
33. """
34. session.query(Users).filter(Users.id > 2).delete()
35. session.commit()
36. """
37. # ################ 修改 ################
38. """
39. session.query(Users).filter(Users.id > 0).update({"name" : "099"})
40. session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
41. session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
42. session.commit()
43. """
44. # ################ 查询 ################
45. """
46. r1 = session.query(Users).all()
47. r2 = session.query(Users.name.label('xx'), Users.age).all()
48. r3 = session.query(Users).filter(Users.name == "alex").all()
49. r4 = session.query(Users).filter_by(name='alex').all()
50. r5 = session.query(Users).filter_by(name='alex').first()
51. r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
52. r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()
53. """
56. session.close()
复制代码
原生SQL语句
1. #　条件
2. ret = session.query(Users).filter_by(name='alex').all()
3. ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
4. ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
5. ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
6. ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
7. ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
8. from sqlalchemy import and_, or_
9. ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
10. ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
11. ret = session.query(Users).filter(
12. or_(
13. Users.id < 2,
14. and_(Users.name == 'eric', Users.id > 3),
15. Users.extra != ""
16. )).all()
19. # 通配符
20. ret = session.query(Users).filter(Users.name.like('e%')).all()
21. ret = session.query(Users).filter(~Users.name.like('e%')).all()
23. # 限制
24. ret = session.query(Users)[1:2]
26. # 排序
27. ret = session.query(Users).order_by(Users.name.desc()).all()
28. ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
30. # 分组
31. from sqlalchemy.sql import func
33. ret = session.query(Users).group_by(Users.extra).all()
34. ret = session.query(
35. func.max(Users.id),
36. func.sum(Users.id),
37. func.min(Users.id)).group_by(Users.name).all()
39. ret = session.query(
40. func.max(Users.id),
41. func.sum(Users.id),
42. func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
44. # 连表
46. ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
48. ret = session.query(Person).join(Favor).all()
50. ret = session.query(Person).join(Favor, isouter=True).all()
53. # 组合
54. q1 = session.query(Users.name).filter(Users.id > 2)
55. q2 = session.query(Favor.caption).filter(Favor.nid < 2)
56. ret = q1.union(q2).all()
58. q1 = session.query(Users.name).filter(Users.id > 2)
59. q2 = session.query(Favor.caption).filter(Favor.nid < 2)
60. ret = q1.union_all(q2).all()
复制代码
基于relationship操作ForeignKey
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import time
4. import threading
6. from sqlalchemy.ext.declarative import declarative_base
7. from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
8. from sqlalchemy.orm import sessionmaker, relationship
9. from sqlalchemy import create_engine
10. from sqlalchemy.sql import text
11. from sqlalchemy.engine.result import ResultProxy
12. from db import Users, Hosts
14. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
15. Session = sessionmaker(bind=engine)
17. session = Session()
19. # 查询
20. # cursor = session.execute('select * from users')
21. # result = cursor.fetchall()
23. # 添加
24. cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
25. session.commit()
26. print(cursor.lastrowid)
28. session.close()
复制代码
基于relationship操作m2m
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import time
4. import threading
6. from sqlalchemy.ext.declarative import declarative_base
7. from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
8. from sqlalchemy.orm import sessionmaker, relationship
9. from sqlalchemy import create_engine
10. from sqlalchemy.sql import text
11. from sqlalchemy.engine.result import ResultProxy
12. from db import Users, Hosts, Hobby, Person
14. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
15. Session = sessionmaker(bind=engine)
16. session = Session()
17. # 添加
18. """
19. session.add_all([
20. Hobby(caption='乒乓球'),
21. Hobby(caption='羽毛球'),
22. Person(name='张三', hobby_id=3),
23. Person(name='李四', hobby_id=4),
24. ])
26. person = Person(name='张九', hobby=Hobby(caption='姑娘'))
27. session.add(person)
29. hb = Hobby(caption='人妖')
30. hb.pers = [Person(name='文飞'), Person(name='博雅')]
31. session.add(hb)
33. session.commit()
34. """
36. # 使用relationship正向查询
37. """
38. v = session.query(Person).first()
39. print(v.name)
40. print(v.hobby.caption)
41. """
43. # 使用relationship反向查询
44. """
45. v = session.query(Hobby).first()
46. print(v.caption)
47. print(v.pers)
48. """
50. session.close()
复制代码
其他
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import time
4. import threading
6. from sqlalchemy.ext.declarative import declarative_base
7. from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
8. from sqlalchemy.orm import sessionmaker, relationship
9. from sqlalchemy import create_engine
10. from sqlalchemy.sql import text
11. from sqlalchemy.engine.result import ResultProxy
12. from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group
14. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
15. Session = sessionmaker(bind=engine)
16. session = Session()
17. # 添加
18. """
19. session.add_all([
20. Server(hostname='c1.com'),
21. Server(hostname='c2.com'),
22. Group(name='A组'),
23. Group(name='B组'),
24. ])
25. session.commit()
27. s2g = Server2Group(server_id=1, group_id=1)
28. session.add(s2g)
29. session.commit()
32. gp = Group(name='C组')
33. gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
34. session.add(gp)
35. session.commit()
38. ser = Server(hostname='c6.com')
39. ser.groups = [Group(name='F组'),Group(name='G组')]
40. session.add(ser)
41. session.commit()
42. """
45. # 使用relationship正向查询
46. """
47. v = session.query(Group).first()
48. print(v.name)
49. print(v.servers)
50. """
52. # 使用relationship反向查询
53. """
54. v = session.query(Server).first()
55. print(v.hostname)
56. print(v.groups)
57. """
60. session.close()
复制代码

14.flask武装

Flask_SQLAlchemy
1. #!/usr/bin/env python
2. # -*- coding:utf-8 -*-
3. import time
4. import threading
6. from sqlalchemy.ext.declarative import declarative_base
7. from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
8. from sqlalchemy.orm import sessionmaker, relationship
9. from sqlalchemy import create_engine
10. from sqlalchemy.sql import text, func
11. from sqlalchemy.engine.result import ResultProxy
12. from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group
14. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
15. Session = sessionmaker(bind=engine)
16. session = Session()
18. # 关联子查询
19. subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
20. result = session.query(Group.name, subqry)
21. """
22. SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid
23. FROM server
24. WHERE server.id = `group`.id) AS anon_1
25. FROM `group`
26. """
29. # 原生SQL
30. """
31. # 查询
32. cursor = session.execute('select * from users')
33. result = cursor.fetchall()
35. # 添加
36. cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
37. session.commit()
38. print(cursor.lastrowid)
39. """
41. session.close()
复制代码
flask-script
1. a. 下载安装
2. pip3 install flask-sqlalchemy
3. b. chun.__init__.py
4. 导入并实例化SQLAlchemy
5. from flask_sqlalchemy import SQLAlchemy
6. db = SQLAlchemy()
8. 注意事项：
9. - 必须在导入蓝图之前
10. - 必须导入models.py
11. c. 初始化
12. db.init_app(app)
14. d. 在配置文件中写入配置
15. # ##### SQLALchemy配置文件 #####
16. SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123456@127.0.0.1:3306/s9day122?charset=utf8"
17. SQLALCHEMY_POOL_SIZE = 10
18. SQLALCHEMY_MAX_OVERFLOW = 5
20. e. 创建models.py中的类（对应数据库表）
21. chun/models.py
22. from sqlalchemy.ext.declarative import declarative_base
23. from sqlalchemy import Column
24. from sqlalchemy import Integer,String,Text,Date,DateTime
25. from sqlalchemy import create_engine
26. from chun import db
29. class Users(db.Model):
30. __tablename__ = 'users'
32. id = Column(Integer, primary_key=True)
33. name = Column(String(32), index=True, nullable=False)
34. depart_id = Column(Integer)
36. f. 生成表（使用app上下文）
37. from chun import db,create_app
39. app = create_app()
40. app_ctx = app.app_context() # app_ctx = app/g
41. with app_ctx: # __enter__,通过LocalStack放入Local中
42. db.create_all() # 调用LocalStack放入Local中获取app，再去app中获取配置
45. g. 基于ORM对数据库进行操作。
46. from flask import Blueprint
47. from chun import db
48. from chun import models
49. us = Blueprint('us',__name__)
52. @us.route('/index')
53. def index():
54. # 使用SQLAlchemy在数据库中插入一条数据
55. # db.session.add(models.Users(name='高件套',depart_id=1))
56. # db.session.commit()
57. # db.session.remove()
58. result = db.session.query(models.Users).all()
59. print(result)
60. db.session.remove()
62. return 'Index'
复制代码
flask-migrate
1. install the module "pip3 install flask-script "
3. a. 增加 runserver
4. from chun import create_app
5. from flask_script import Manager
8. app = create_app()
9. manager = Manager(app)
11. if __name__ == '__main__':
12. # app.run()
13. manager.run()
15. b. 位置传参
16. from chun import create_app
17. from flask_script import Manager
20. app = create_app()
21. manager = Manager(app)
23. @manager.command
24. def custom(arg):
25. """
26. 自定义命令
27. python manage.py custom 123
28. :param arg:
29. :return:
30. """
31. print(arg)
34. if __name__ == '__main__':
35. # app.run()
36. manager.run()
37. c. 关键字传参
38. from chun import create_app
39. from flask_script import Manager
42. app = create_app()
43. manager = Manager(app)
45. @manager.option('-n', '--name', dest='name')
46. @manager.option('-u', '--url', dest='url')
47. def cmd(name, url):
48. """
49. 自定义命令
50. 执行： python manage.py cmd -n wupeiqi -u http://www.oldboyedu.com
51. :param name:
52. :param url:
53. :return:
54. """
55. print(name, url)
58. if __name__ == '__main__':
59. # app.run()
60. manager.run()
复制代码
找到项目使用的所有组件和版本。
1. install the module "pip3 install flask-migrate"
2. flask-migrate rely on flask-script.
5. #!/usr/bin/env python
6. # -*- coding:utf-8 -*-
8. from sansa import create_app
9. from sansa import db
11. from flask_script import Manager
12. from flask_migrate import Migrate, MigrateCommand
14. app = create_app()
15. manager = Manager(app)
16. Migrate(app, db)
18. """
19. # 数据库迁移命名
20. python manage.py db init
21. python manage.py db migrate
22. python manage.py db upgrade
23. """
24. manager.add_command('db', MigrateCommand)
27. if __name__ == '__main__':
28. manager.run()
29. # app.run()
复制代码
python虚拟环境
1. install the module "pip install pipreqs"
2. # 终端输入后就可以找到所有的依赖
3. pipreqs ./ --encoding=utf-8
复制代码

来源:https://www.cnblogs.com/InvincibleGrass/p/17562878.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】我们会及时删除侵权内容，谢谢合作！