flask知识梳理

2019年11月19日 阅读数:66
这篇文章主要向大家介绍flask知识梳理,主要内容包括基础应用、实用技巧、原理机制等方面,希望对大家有所帮助。

Flask

0.Flask简介

Flask是一个基于Python开发而且依赖jinja2模板和Werkzeug WSGI服务的一个微型框架,对于Werkzeug本质是Socket服务端,其用于接收http请求并对请求进行预处理,而后触发Flask框架,开发人员基于Flask框架提供的功能对请求进行相应的处理,并返回给用户,若是要返回给用户复杂的内容时,须要借助jinja2模板来实现对模板的处理,即:将模板和数据进行渲染,将渲染后的字符串返回给用户浏览器。html

“微”(micro) 并不表示你须要把整个 Web 应用塞进单个 Python 文件(虽然确实能够 ),也不意味着 Flask 在功能上有所欠缺。微框架中的“微”意味着 Flask 旨在保持核心简单而易于扩展。Flask 不会替你作出太多决策——好比使用何种数据库。而那些 Flask 所选择的——好比使用何种模板引擎——则很容易替换。除此以外的一切都由可由你掌握。如此,Flask 能够与您珠联璧合。html5

默认状况下,Flask 不包含数据库抽象层、表单验证,或是其它任何已有多种库能够胜任的功能。然而,Flask 支持用扩展来给应用添加这些功能,如同是 Flask 自己实现的同样。众多的扩展提供了数据库集成、表单验证、上传处理、各类各样的开放认证技术等功能。Flask 也许是“微小”的,但它已准备好在需求繁杂的生产环境中投入使用python

1.安装

pip3 install flaskmysql

2.werkzeug简介

Werkzeug是一个WSGI工具包,他能够做为一个Web框架的底层库。这里稍微说一下, werkzeug 不是一个web服务器,也不是一个web框架,而是一个工具包,官方的介绍说是一个 WSGI 工具包,它能够做为一个 Web 框架的底层库,由于它封装好了不少 Web 框架的东西,例如 Request,Response 等等 web

代码示例:正则表达式

from werkzeug.wrappers import Request, Response

@Request.application
def hello(request):
   return Response('Hello World!')

if __name__ == '__main__':
   from werkzeug.serving import run_simple
   run_simple('localhost', 4000, hello)

3.flask快速使用


from flask import Flask
# 实例化产生一个Flask对象
app = Flask(__name__)
# 将 '/'和视图函数hello_workd的对应关系添加到路由中
@app.route('/') # 1. v=app.route('/') 2. v(hello_world)
def hello_world():
   return 'Hello World!'

if __name__ == '__main__':
   app.run() # 最终调用了run_simple()

案例:登陆,显示用户信息

main.pyredis


from flask import Flask,render_template,request,redirect,session,url_for
app = Flask(__name__)
app.debug = True
app.secret_key = 'sdfsdfsdfsdf'

USERS = {
   1:{'name':'张三','age':18,'gender':'男','text':"道路千万条"},
   2:{'name':'李四','age':28,'gender':'男','text':"安全第一条"},
   3:{'name':'王五','age':18,'gender':'女','text':"行车不规范"},
}

@app.route('/detail/<int:nid>',methods=['GET'])
def detail(nid):
   user = session.get('user_info')
   if not user:
       return redirect('/login')

   info = USERS.get(nid)
   return render_template('detail.html',info=info)


@app.route('/index',methods=['GET'])
def index():
   user = session.get('user_info')
   if not user:
       # return redirect('/login')
       url = url_for('l1')
       return redirect(url)
   return render_template('index.html',user_dict=USERS)


@app.route('/login',methods=['GET','POST'],endpoint='l1')
def login():
   if request.method == "GET":
       return render_template('login.html')
   else:
       # request.query_string
       user = request.form.get('user')
       pwd = request.form.get('pwd')
       if user == 'lqz' and pwd == '123':
           session['user_info'] = user
           return redirect('http://www.baidu.com')
       return render_template('login.html',error='用户名或密码错误')

if __name__ == '__main__':
   app.run()

detail.htmlsql


<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>
   <h1>详细信息 {{info.name}}</h1>
   <div>
      {{info.text}}
   </div>
</body>
</html>

index.html数据库


<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>
   <h1>用户列表</h1>
   <table>
      {% for k,v in user_dict.items() %}
       <tr>
           <td>{{k}}</td>
           <td>{{v.name}}</td>
           <td>{{v['name']}}</td>
           <td>{{v.get('name')}}</td>
           <td><a href="/detail/{{k}}">查看详细</a></td>
       </tr>
      {% endfor %}
   </table>
</body>
</html>

login.htmldjango


<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>
   <h1>用户登陆</h1>
   <form method="post">
       <input type="text" name="user">
       <input type="text" name="pwd">
       <input type="submit" value="登陆">{{error}}
   </form>
</body>
</html>

4.配置文件

flask中的配置文件是一个flask.config.Config对象(继承字典),默认配置为:


{
       'DEBUG':                                get_debug_flag(default=False),  是否开启Debug模式
       'TESTING':                              False,                          是否开启测试模式
       'PROPAGATE_EXCEPTIONS':                 None,                          
       'PRESERVE_CONTEXT_ON_EXCEPTION':        None,
       'SECRET_KEY':                           None,
       'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),
       'USE_X_SENDFILE':                       False,
       'LOGGER_NAME':                          None,
       'LOGGER_HANDLER_POLICY':               'always',
       'SERVER_NAME':                          None,
       'APPLICATION_ROOT':                     None,
       'SESSION_COOKIE_NAME':                  'session',
       'SESSION_COOKIE_DOMAIN':                None,
       'SESSION_COOKIE_PATH':                  None,
       'SESSION_COOKIE_HTTPONLY':              True,
       'SESSION_COOKIE_SECURE':                False,
       'SESSION_REFRESH_EACH_REQUEST':         True,
       'MAX_CONTENT_LENGTH':                   None,
       'SEND_FILE_MAX_AGE_DEFAULT':            timedelta(hours=12),
       'TRAP_BAD_REQUEST_ERRORS':              False,
       'TRAP_HTTP_EXCEPTIONS':                 False,
       'EXPLAIN_TEMPLATE_LOADING':             False,
       'PREFERRED_URL_SCHEME':                 'http',
       'JSON_AS_ASCII':                        True,
       'JSON_SORT_KEYS':                       True,
       'JSONIFY_PRETTYPRINT_REGULAR':          True,
       'JSONIFY_MIMETYPE':                     'application/json',
       'TEMPLATES_AUTO_RELOAD':                None,
  }

方式一


  app.config['DEBUG'] = True
  PS: 因为Config对象本质上是字典,因此还可使用app.config.update(...)

方式二


#经过py文件配置
app.config.from_pyfile("python文件名称")
如:
settings.py
DEBUG = True

app.config.from_pyfile("settings.py")
#经过环境变量配置
app.config.from_envvar("环境变量名称")
#app.config.from_pyfile(os.environ['YOURAPPLICATION_SETTINGS'])
环境变量的值为python文件名称名称,内部调用from_pyfile方法

app.config.from_json("json文件名称")
JSON文件名称,必须是json格式,由于内部会执行json.loads

app.config.from_mapping({'DEBUG': True})
字典格式

app.config.from_object("python类或类的路径")

app.config.from_object('pro_flask.settings.TestingConfig')

settings.py


class Config(object):
   DEBUG = False
   TESTING = False
   DATABASE_URI = 'sqlite://:memory:'


class ProductionConfig(Config):
   DATABASE_URI = 'mysql://user@localhost/foo'


class DevelopmentConfig(Config):
   DEBUG = True


class TestingConfig(Config):
   TESTING = True


PS: 从sys.path中已经存在路径开始写

PS: settings.py文件默认路径要放在程序root_path目录,若是instance_relative_config为True,则就是instance_path目录(Flask对象init方法的参数)

5.路由系统

典型写法


@app.route('/detail/<int:nid>',methods=['GET'],endpoint='detail')

默认转换器


DEFAULT_CONVERTERS = {
   'default':          UnicodeConverter,
   'string':           UnicodeConverter,
   'any':              AnyConverter,
   'path':             PathConverter,
   'int':              IntegerConverter,
   'float':            FloatConverter,
   'uuid':             UUIDConverter,
}

路由系统本质


"""
1. decorator = app.route('/',methods=['GET','POST'],endpoint='n1')
  def route(self, rule, **options):
      # app对象
      # rule= /
      # options = {methods=['GET','POST'],endpoint='n1'}
      def decorator(f):
          endpoint = options.pop('endpoint', None)
          self.add_url_rule(rule, endpoint, f, **options)
          return f
      return decorator
2. @decorator
  decorator(index)
"""
#同理
def login():
   return '登陆'
app.add_url_rule('/login', 'n2', login, methods=['GET',"POST"])
#与django路由相似
#django与flask路由:flask路由基于装饰器,本质是基于:add_url_rule
#add_url_rule 源码中,endpoint若是为空,endpoint = _endpoint_from_view_func(view_func),最终取view_func.__name__(函数名)

CBV(源码分析)


def auth(func):
   def inner(*args, **kwargs):
       print('before')
       result = func(*args, **kwargs)
       print('after')
       return result

   return inner

class IndexView(views.View):
   methods = ['GET']
   decorators = [auth, ]

   def dispatch_request(self):
       print('Index')
       return 'Index!'

app.add_url_rule('/index', view_func=IndexView.as_view(name='index'))  # name=endpoint
#或者,一般用此方式
 class IndexView(views.MethodView):
           methods = ['GET']
           decorators = [auth, ]

           def get(self):
               return 'Index.GET'

           def post(self):
               return 'Index.POST'
       app.add_url_rule('/index', view_func=IndexView.as_view(name='index'))  # name=endpoint

app.add_url_rule参数


@app.route和app.add_url_rule参数:
rule, URL规则
view_func, 视图函数名称
defaults = None, 默认值, 当URL中无参数,函数须要参数时,使用defaults = {'k': 'v'}
为函数提供参数
endpoint = None, 名称,用于反向生成URL,即: url_for('名称')
methods = None, 容许的请求方式,如:["GET", "POST"]
#对URL最后的 / 符号是否严格要求
strict_slashes = None
   '''
      @app.route('/index', strict_slashes=False)
      #访问http://www.xx.com/index/ 或http://www.xx.com/index都可
      @app.route('/index', strict_slashes=True)
      #仅访问http://www.xx.com/index
  '''
#重定向到指定地址
redirect_to = None,
   '''
      @app.route('/index/<int:nid>', redirect_to='/home/<nid>')
  '''

#子域名访问
subdomain = None,
   '''
  #C:\Windows\System32\drivers\etc\hosts
  127.0.0.1       www.liuqingzheng.com
127.0.0.1       admin.liuqingzheng.com
127.0.0.1       buy.liuqingzheng.com
   
  from flask import Flask, views, url_for
  app = Flask(import_name=__name__)
  app.config['SERVER_NAME'] = 'liuqingzheng.com:5000'
  @app.route("/", subdomain="admin")
  def static_index():
      """Flask supports static subdomains
      This is available at static.your-domain.tld"""
      return "static.your-domain.tld"
  #能够传入任意的字符串,如传入的字符串为aa,显示为 aa.liuqingzheng.com
  @app.route("/dynamic", subdomain="<username>")
  def username_index(username):
      """Dynamic subdomains are also supported
      Try going to user1.your-domain.tld/dynamic"""
      return username + ".your-domain.tld"
  if __name__ == '__main__':
      app.run()
       
  访问:
  http://www.liuqingzheng.com:5000/dynamic
  http://admin.liuqingzheng.com:5000/dynamic
  http://buy.liuqingzheng.com:5000/dynamic
  '''

支持正则


#1 写类,继承BaseConverter
#2 注册:app.url_map.converters['regex'] = RegexConverter
# 3 使用:@app.route('/index/<regex("\d+"):nid>') 正则表达式会看成第二个参数传递到类中
from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter

app = Flask(import_name=__name__)

class RegexConverter(BaseConverter):
   """
  自定义URL匹配正则表达式
  """
   def __init__(self, map, regex):
       super(RegexConverter, self).__init__(map)
       self.regex = regex

   def to_python(self, value):
       """
      路由匹配时,匹配成功后传递给视图函数中参数的值
      """
       return int(value)

   def to_url(self, value):
       """
      使用url_for反向生成URL时,传递的参数通过该方法处理,返回的值用于生成URL中的参数
      """
       val = super(RegexConverter, self).to_url(value)
       return val
# 添加到flask中
app.url_map.converters['regex'] = RegexConverter
@app.route('/index/<regex("\d+"):nid>')
def index(nid):
   print(url_for('index', nid='888'))
   return 'Index'

if __name__ == '__main__':
   app.run()

6.模版

比django中多能够加括号,执行函数,传参数


from flask import Flask,render_template,Markup,jsonify,make_response
app = Flask(__name__)

def func1(arg):
   return Markup("<input type='text' value='%s' />" %(arg,))
@app.route('/')
def index():
   return render_template('index.html',ff = func1)

if __name__ == '__main__':
   app.run()

index.html


<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>

  {{ff('六五')}}
{{ff('六五')|safe}}

</body>
</html>

注意:

1.Markup等价django的mark_safe ,

2.extends,include如出一辙

7.请求响应


  from flask import Flask
   from flask import request
   from flask import render_template
   from flask import redirect
   from flask import make_response

   app = Flask(__name__)


   @app.route('/login.html', methods=['GET', "POST"])
   def login():

       # 请求相关信息
       # request.method
       # request.args #至关于django中的request.GET
       # request.form #至关于django中的request.POST
       # request.values
       # request.cookies
       # request.headers
       # request.path
       # request.full_path
       # request.script_root
       # request.url
       # request.base_url
       # request.url_root
       # request.host_url
       # request.host
       # request.files
       # obj = request.files['the_file_name']
       # obj.save('/var/www/uploads/' + secure_filename(f.filename))

       # 响应相关信息
       # return "字符串"
       # return render_template('html模板路径',**{})
       # return redirect('/index.html')
       #return jsonify({'k1':'v1'})

       # response = make_response(render_template('index.html'))
       # response是flask.wrappers.Response类型
       # response.delete_cookie('key')
       # response.set_cookie('key', 'value')
       # response.headers['X-Something'] = 'A value'
       # return response
       return "内容"

   if __name__ == '__main__':
       app.run()

8.session

除请求对象以外,还有一个 session 对象。它容许你在不一样请求间存储特定用户的信息。它是在 Cookies 的基础上实现的,而且对 Cookies 进行密钥签名要使用会话,须要设置一个密钥。 (app.session_interface对象)

app.secret_key='xxxxx'
设置:session['username'] 'xxx'
取值:username=session['username']
删除:session.pop('username', None)
   del session['username']

9.闪现(message)


-设置:flash('aaa')
-取值:get_flashed_message()
-假设在a页面操做出错,跳转到b页面,在b页面显示a页面的错误信息

示例:


from flask import Flask,flash,get_flashed_messages,request,redirect

app = Flask(__name__)
app.secret_key = 'asdfasdf'


@app.route('/index')
def index():
   # 从某个地方获取设置过的全部值,并清除。
   val = request.args.get('v')
   if val == 'oldboy':
       return 'Hello World!'
   flash('超时错误',category="x1")
   return "ssdsdsdfsd"
   # return redirect('/error')


@app.route('/error')
def error():
   """
  展现错误信息
  :return:
  """
   data = get_flashed_messages(category_filter=['x1'])
   if data:
       msg = data[0]
   else:
       msg = "..."
   return "错误信息:%s" %(msg,)

if __name__ == '__main__':
   app.run()

10.请求扩展

1 before_request

类比django中间件中的process_request,在请求收到以前绑定一个函数作一些事情


#基于它作用户登陆认证
@app.before_request
def process_request(*args,**kwargs):
   if not request.path == '/login':
     user = session.get('user_info')
     if not user:
       return redirect('/login')

2 after_request

类比django中间件中的process_response,每个请求以后绑定一个函数,若是请求没有异常


@app.after_request
def process_response1(response):
   print('process_response1 走了')
   return response

3 before_first_request

第一次请求时,跟浏览器无关


@app.before_first_request
def first():
   pass

4 teardown_request

每个请求以后绑定一个函数,即便遇到了异常(每一次请求不管如何都会通过此函数)


@app.teardown_request
def ter(e):
   pass

5 errorhandler

路径不存在时404,服务器内部错误500

@app.errorhandler(404)
def error_404(err):
  print(err)
  return "404错误了"

6 template_global

标签


@app.template_global()
def add(a1, a2):
   return a1 + a2
#{{add(1,2)}}

7 template_filter

过滤器


@app.template_filter()
def add(a1, a2, a3):
   return a1 + a2 + a3
#{{ 1|add(2,3)}}

总结:

1 重点掌握before_request和after_request,

2 注意有多个的状况,执行顺序

3 before_request请求拦截后(也就是有return值),response全部都执行

11 中间件


from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
   return 'Hello World!'
# 模拟中间件
class MyMiddle(object):
   def __init__(self,old_wsgi_app):
       self.old_wsgi_app = old_wsgi_app

   def __call__(self,  environ, start_response):
       print('开始以前')
       ret = self.old_wsgi_app(environ, start_response)
       print('结束以后')
       return ret

if __name__ == '__main__':
   #把原来的wsgi_app替换为自定义的
   app.wsgi_app = MyMiddle(app.wsgi_app)
   app.run()

12.蓝图

对程序进行目录结构划分

不使用蓝图,本身分文件

目录结构:


-templates
-views
-__init__.py
   -user.py
   -order.py
-app.py

app.py


from views import app
if __name__ == '__main__':
   app.run()

init.py


from flask import Flask,request
app = Flask(__name__)
#不导入这个不行
from . import account
from . import order
from . import user

user.py


from . import app
@app.route('/user')
def user():
   return 'user'

order.py


from . import app
@app.route('/order')
def order():
   return 'order'

使用蓝图之中小型系统

详见代码:pro_flask_简单应用程序目录示例.zip

目录结构:


-flask_pro
-flask_test
  -__init__.py
  -static
       -templates
       -views
      -order.py
           -user.py
    -manage.py
       

_init.py


from flask import  Flask
app=Flask(__name__)
from flask_test.views import user
from flask_test.views import order
#注册蓝图
app.register_blueprint(user.us)
app.register_blueprint(order.ord)

manage.py


from flask_test import  app
if __name__ == '__main__':
   app.run(port=8008)

user.py


from flask import Blueprint
us=Blueprint('user',__name__)

@us.route('/login')
def login():
   return 'login'

order.py


from flask import Blueprint
ord=Blueprint('order',__name__)

@ord.route('/test')
def test():
   return 'order test'

使用蓝图之大型系统

详见代码:pro_flask_大型应用目录示例.zip

总结:

1 xxx = Blueprint('account', name,url_prefix='/xxx') :蓝图URL前缀,表示url的前缀,在该蓝图下全部url都加前缀

2 xxx = Blueprint('account', name,url_prefix='/xxx',template_folder='tpls'):给当前蓝图单独使用templates,向上查找,当前找不到,会找总templates

3 蓝图的befort_request,对当前蓝图有效

4 大型项目,能够模拟出相似于django中app的概念

13.请求上下文源码分析


第一阶段:将ctx(request,session)放到Local对象上
 
第二阶段:视图函数导入:request/session
request.method
-LocalProxy对象.method,执行getattr方法,getattr(self._get_current_object(), name)
-self._get_current_object()返回return self.__local(),self.__local(),在LocakProxy实例化的时候,object.__setattr__(self, '_LocalProxy__local', local),此处local就是:partial(_lookup_req_object, 'request')

-def _lookup_req_object(name):
top = _request_ctx_stack.top #_request_ctx_stack 就是LocalStack()对象,top方法把ctx取出来
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name)#获取ctx中的request或session对象

第三阶段:请求处理完毕
- 获取session并保存到cookie
- 将ctx删除

程序运行,两个LocalStack()对象,一个里面放request和session,另外一个放g和current_app

14.g对象

专门用来存储用户信息的g对象,g的全称的为global

g对象在一次请求中的全部的代码的地方,都是可使用的

g对象和session的区别


session对象是能够跨request的,只要session还未失效,不一样的request的请求会获取到同一个session,可是g对象不是,g对象不须要管过时时间,请求一次就g对象就改变了一次,或者从新赋值了一次

15.flask-session

做用:将默认保存的签名cookie中的值 保存到 redis/memcached/file/Mongodb/SQLAlchemy

安装:pip3 install flask-session

使用1:

from flask import Flask,session
from flask_session import RedisSessionInterface
import redis
app = Flask(__name__)
conn=redis.Redis(host='127.0.0.1',port=6379)
#use_signer是否对key签名
app.session_interface=RedisSessionInterface(conn,key_prefix='lqz')
@app.route('/')
def hello_world():
   session['name']='lqz'
   return 'Hello World!'

if __name__ == '__main__':
   app.run()

使用2:


from redis import Redis
from flask.ext.session import Session
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379')
Session(app)

问题:设置cookie时,如何设定关闭浏览器则cookie失效。


response.set_cookie('k','v',exipre=None)#这样设置便可
#在session中设置
app.session_interface=RedisSessionInterface(conn,key_prefix='lqz',permanent=False)
#通常不用,咱们通常都设置超时时间,多长时间后失效

问题:cookie默认超时时间是多少?如何设置超时时间


#源码expires = self.get_expiration_time(app, session)
'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),#这个配置文件控制

 

 

16.数据库链接池

pymsql连接数据库


import pymysql

conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123456', db='s8day127db')
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
# cursor.execute("select id,name from users where name=%s and pwd=%s",['lqz','123',])
cursor.execute("select id,name from users where name=%(user)s and pwd=%(pwd)s",{'user':'lqz','pwd':'123'})
obj = cursor.fetchone()
conn.commit()
cursor.close()
conn.close()

print(obj)

数据库链接池版

setting.py


from datetime import timedelta
from redis import Redis
import pymysql
from DBUtils.PooledDB import PooledDB, SharedDBConnection

class Config(object):
   DEBUG = True
   SECRET_KEY = "umsuldfsdflskjdf"
   PERMANENT_SESSION_LIFETIME = timedelta(minutes=20)
   SESSION_REFRESH_EACH_REQUEST= True
   SESSION_TYPE = "redis"
   PYMYSQL_POOL = PooledDB(
       creator=pymysql,  # 使用连接数据库的模块
       maxconnections=6,  # 链接池容许的最大链接数,0和None表示不限制链接数
       mincached=2,  # 初始化时,连接池中至少建立的空闲的连接,0表示不建立
       maxcached=5,  # 连接池中最多闲置的连接,0和None不限制
       maxshared=3,
       # 连接池中最多共享的连接数量,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='s8day127db',
       charset='utf8'
  )

class ProductionConfig(Config):
   SESSION_REDIS = Redis(host='192.168.0.94', port='6379')



class DevelopmentConfig(Config):
   SESSION_REDIS = Redis(host='127.0.0.1', port='6379')


class TestingConfig(Config):
   pass

utils/sql.py


import pymysql
from settings import Config
class SQLHelper(object):

   @staticmethod
   def open(cursor):
       POOL = Config.PYMYSQL_POOL
       conn = POOL.connection()
       cursor = conn.cursor(cursor=cursor)
       return conn,cursor

   @staticmethod
   def close(conn,cursor):
       conn.commit()
       cursor.close()
       conn.close()

   @classmethod
   def fetch_one(cls,sql,args,cursor =pymysql.cursors.DictCursor):
       conn,cursor = cls.open(cursor)
       cursor.execute(sql, args)
       obj = cursor.fetchone()
       cls.close(conn,cursor)
       return obj

   @classmethod
   def fetch_all(cls,sql, args,cursor =pymysql.cursors.DictCursor):
       conn, cursor = cls.open(cursor)
       cursor.execute(sql, args)
       obj = cursor.fetchall()
       cls.close(conn, cursor)
       return obj

使用:


obj = SQLHelper.fetch_one("select id,name from users where name=%(user)s and pwd=%(pwd)s", form.data)

 

17.wtforms

安装:pip3 install wtforms

使用1:


from flask import Flask, render_template, request, redirect
from wtforms import Form
from wtforms.fields import simple
from wtforms import validators
from wtforms import widgets

app = Flask(__name__, template_folder='templates')

app.debug = True


class LoginForm(Form):
   # 字段(内部包含正则表达式)
   name = simple.StringField(
       label='用户名',
       validators=[
           validators.DataRequired(message='用户名不能为空.'),
           validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d')
      ],
       widget=widgets.TextInput(), # 页面上显示的插件
       render_kw={'class': 'form-control'}

  )
   # 字段(内部包含正则表达式)
   pwd = simple.PasswordField(
       label='密码',
       validators=[
           validators.DataRequired(message='密码不能为空.'),
           validators.Length(min=8, message='用户名长度必须大于%(min)d'),
           validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[$@$!%*?&])[A-Za-z\d$@$!%*?&]{8,}",
                             message='密码至少8个字符,至少1个大写字母,1个小写字母,1个数字和1个特殊字符')

      ],
       widget=widgets.PasswordInput(),
       render_kw={'class': 'form-control'}
  )



@app.route('/login', methods=['GET', 'POST'])
def login():
   if request.method == 'GET':
       form = LoginForm()
       return render_template('login.html', form=form)
   else:
       form = LoginForm(formdata=request.form)
       if form.validate():
           print('用户提交数据经过格式验证,提交的值为:', form.data)
       else:
           print(form.errors)
       return render_template('login.html', form=form)

if __name__ == '__main__':
   app.run()

login.html


<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>
<h1>登陆</h1>
<form method="post">
   <p>{{form.name.label}} {{form.name}} {{form.name.errors[0] }}</p>

   <p>{{form.pwd.label}} {{form.pwd}} {{form.pwd.errors[0] }}</p>
   <input type="submit" value="提交">
</form>
</body>
</html>

使用2:


from flask import Flask, render_template, request, redirect
from wtforms import Form
from wtforms.fields import core
from wtforms.fields import html5
from wtforms.fields import simple
from wtforms import validators
from wtforms import widgets

app = Flask(__name__, template_folder='templates')
app.debug = True



class RegisterForm(Form):
   name = simple.StringField(
       label='用户名',
       validators=[
           validators.DataRequired()
      ],
       widget=widgets.TextInput(),
       render_kw={'class': 'form-control'},
       default='alex'
  )

   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 # “1” “2”
    )
   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,
       default=[1, 2]
  )

   def __init__(self, *args, **kwargs):
       super(RegisterForm, self).__init__(*args, **kwargs)
       self.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球'))

   def validate_pwd_confirm(self, field):
       """
      自定义pwd_confirm字段规则,例:与pwd字段是否一致
      :param field:
      :return:
      """
       # 最开始初始化时,self.data中已经有全部的值

       if field.data != self.data['pwd']:
           # raise validators.ValidationError("密码不一致") # 继续后续验证
           raise validators.StopValidation("密码不一致")  # 再也不继续后续验证


@app.route('/register', methods=['GET', 'POST'])
def register():
   if request.method == 'GET':
       form = RegisterForm(data={'gender': 2,'hobby':[1,]}) # initial
       return render_template('register.html', form=form)
   else:
       form = RegisterForm(formdata=request.form)
       if form.validate():
           print('用户提交数据经过格式验证,提交的值为:', form.data)
       else:
           print(form.errors)
       return render_template('register.html', form=form)



if __name__ == '__main__':
   app.run()

<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>
<h1>用户注册</h1>
<form method="post" novalidate style="padding:0 50px">
  {% for field in form %}
   <p>{{field.label}}: {{field}} {{field.errors[0] }}</p>
  {% endfor %}
   <input type="submit" value="提交">
</form>
</body>
</html>

 

18.信号

Flask框架中的信号基于blinker,其主要就是让开发者但是在flask请求过程当中定制一些用户行为

安装:pip3 install blinker

内置信号:


request_started = _signals.signal('request-started')                # 请求到来前执行
request_finished = _signals.signal('request-finished')              # 请求结束后执行

before_render_template = _signals.signal('before-render-template')  # 模板渲染前执行
template_rendered = _signals.signal('template-rendered')            # 模板渲染后执行

got_request_exception = _signals.signal('got-request-exception')    # 请求执行出现异常时执行

request_tearing_down = _signals.signal('request-tearing-down')      # 请求执行完毕后自动执行(不管成功与否)
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行(不管成功与否)

appcontext_pushed = _signals.signal('appcontext-pushed')            # 应用上下文push时执行
appcontext_popped = _signals.signal('appcontext-popped')            # 应用上下文pop时执行
message_flashed = _signals.signal('message-flashed')                # 调用flask在其中添加数据时,自动触发

使用信号:


from flask import Flask,signals,render_template

app = Flask(__name__)

# 往信号中注册函数
def func(*args,**kwargs):
   print('触发型号',args,kwargs)
signals.request_started.connect(func)

# 触发信号: signals.request_started.send()
@app.before_first_request
def before_first1(*args,**kwargs):
   pass
@app.before_first_request
def before_first2(*args,**kwargs):
   pass

@app.before_request
def before_first3(*args,**kwargs):
   pass

@app.route('/',methods=['GET',"POST"])
def index():
   print('视图')
   return render_template('index.html')


if __name__ == '__main__':
   app.wsgi_app
   app.run()

一个流程中的信号触发点(了解)


a. before_first_request
b. 触发 request_started 信号
c. before_request
d. 模板渲染
渲染前的信号 before_render_template.send(app, template=template, context=context)
rv = template.render(context) # 模板渲染
渲染后的信号 template_rendered.send(app, template=template, context=context)
e. after_request
f. session.save_session()
g. 触发 request_finished信号
若是上述过程出错:
触发错误处理信号 got_request_exception.send(self, exception=e)

h. 触发信号 request_tearing_down

自定义信号(了解):


from flask import Flask, current_app, flash, render_template
from flask.signals import _signals
app = Flask(import_name=__name__)

# 自定义信号
xxxxx = _signals.signal('xxxxx')

def func(sender, *args, **kwargs):
   print(sender)
# 自定义信号中注册函数
xxxxx.connect(func)
@app.route("/x")
def index():
   # 触发信号
   xxxxx.send('123123', k1='v1')
   return 'Index'

if __name__ == '__main__':
   app.run()

 

19.多app应用


from werkzeug.wsgi import DispatcherMiddleware
from werkzeug.serving import run_simple
from flask import Flask, current_app
app1 = Flask('app01')
app2 = Flask('app02')

@app1.route('/index')
def index():
   return "app01"

@app2.route('/index2')
def index2():
   return "app2"

# http://www.oldboyedu.com/index
# http://www.oldboyedu.com/sec/index2
dm = DispatcherMiddleware(app1, {
   '/sec': app2,
})

if __name__ == "__main__":
   run_simple('localhost', 5000, dm)

20.flask-script

用于实现相似于django中 python3 manage.py runserver ...相似的命令

安装:pip3 install flask-script

使用


from flask_script import Manager
app = Flask(__name__)
manager=Manager(app)
...
if __name__ == '__main__':
   manager.run()
#之后在执行,直接:python3 manage.py runserver
#python3 manage.py runserver --help

自定制命令

@manager.command
def custom(arg):
   """
  自定义命令
  python manage.py custom 123
  :param arg:
  :return:
  """
   print(arg)


@manager.option('-n', '--name', dest='name')
#@manager.option('-u', '--url', dest='url')
def cmd(name, url):
   """
  自定义命令(-n也能够写成--name)
  执行: python manage.py cmd -n lqz -u http://www.oldboyedu.com
  执行: python manage.py cmd --name lqz --url http://www.oldboyedu.com
  :param name:
  :param url:
  :return:
  """
   print(name, url)
#有什么用?
#把excel的数据导入数据库,定制个命令,去执行

 

SQLAlchemy

1.介绍

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,从而实现对数据库的操做,如:


MySQL-Python
   mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
pymysql
   mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
MySQL-Connector
   mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
cx_Oracle
   oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
   
更多:http://docs.sqlalchemy.org/en/latest/dialects/index.html

2.简单使用(能建立表,删除表,不能修改表)

修改表:在数据库添加字段,类对应上

1执行原生sql(不经常使用)

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

engine = create_engine(
   "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
   max_overflow=0,  # 超过链接池大小外最多建立的链接
   pool_size=5,  # 链接池大小
   pool_timeout=30,  # 池中没有线程最多等待的时间,不然报错
   pool_recycle=-1  # 多久以后对线程池中的线程进行一次链接的回收(重置)
)
def task(arg):
   conn = engine.raw_connection()
   cursor = conn.cursor()
   cursor.execute(
       "select * from app01_book"
  )
   result = cursor.fetchall()
   print(result)
   cursor.close()
   conn.close()

for i in range(20):
   t = threading.Thread(target=task, args=(i,))
   t.start()

2 orm使用

models.py


import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
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)  # 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, nullable=True)

   __table_args__ = (
       # UniqueConstraint('id', 'name', name='uix_id_name'), #联合惟一
       # Index('ix_id_name', 'name', 'email'), #索引
  )

def init_db():
   """
  根据类建立数据库表
  :return:
  """
   engine = create_engine(
       "mysql+pymysql://root:123456@127.0.0.1:3306/aaa?charset=utf8",
       max_overflow=0,  # 超过链接池大小外最多建立的链接
       pool_size=5,  # 链接池大小
       pool_timeout=30,  # 池中没有线程最多等待的时间,不然报错
       pool_recycle=-1  # 多久以后对线程池中的线程进行一次链接的回收(重置)
  )

   Base.metadata.create_all(engine)

def drop_db():
   """
  根据类删除数据库表
  :return:
  """
   engine = create_engine(
       "mysql+pymysql://root:123456@127.0.0.1:3306/aaa?charset=utf8",
       max_overflow=0,  # 超过链接池大小外最多建立的链接
       pool_size=5,  # 链接池大小
       pool_timeout=30,  # 池中没有线程最多等待的时间,不然报错
       pool_recycle=-1  # 多久以后对线程池中的线程进行一次链接的回收(重置)
  )

   Base.metadata.drop_all(engine)

if __name__ == '__main__':
   # drop_db()
   init_db()

app.py


from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#"mysql+pymysql://root@127.0.0.1:3306/aaa"
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)
Connection = sessionmaker(bind=engine)

# 每次执行数据库操做时,都须要建立一个Connection
con = Connection()

# ############# 执行ORM操做 #############
obj1 = Users(name="lqz")
con.add(obj1)
# 提交事务
con.commit()

# 关闭session,实际上是将链接放回链接池
con.close()

3.一对多关系


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), index=True, nullable=True)
   # hobby指的是tablename而不是类名
   hobby_id = Column(Integer, ForeignKey("hobby.id"))
   
   # 跟数据库无关,不会新增字段,只用于快速链表操做
   # 类名,backref用于反向查询
   hobby=relationship('Hobby',backref='pers')

4.多对多关系


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)
   hostname = Column(String(64), unique=True, nullable=False)
   
   # 与生成表结构无关,仅用于查询方便,放在哪一个单表中均可以
   servers = relationship('Girl', secondary='boy2girl', backref='boys')

5.操做数据表


from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
 
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
 
# 每次执行数据库操做时,都须要建立一个session
session = Session()
 
# ############# 执行ORM操做 #############
obj1 = Users(name="lqz")
session.add(obj1)
 
# 提交事务
session.commit()
# 关闭session
session.close()

6.基于scoped_session实现线程安全


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:123@127.0.0.1:3306/s6", 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="alex1")
session.add(obj1)

# 提交事务
session.commit()
# 关闭session
session.close()

7.基本增删查改


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 db import Users, Hosts

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

# ################ 添加 ################
"""
obj1 = Users(name="wupeiqi")
session.add(obj1)

session.add_all([
  Users(name="lqz"),
  Users(name="egon"),
  Hosts(name="c1.com"),
])
session.commit()
"""

# ################ 删除 ################
"""
session.query(Users).filter(Users.id > 2).delete()
session.commit()
"""
# ################ 修改 ################
"""
#传字典
session.query(Users).filter(Users.id > 0).update({"name" : "lqz"})
#相似于django的F查询
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
session.commit()
"""
# ################ 查询 ################
"""
r1 = session.query(Users).all()
#只取age列,把name重命名为xx
r2 = session.query(Users.name.label('xx'), Users.age).all()
#filter传的是表达式,filter_by传的是参数
r3 = session.query(Users).filter(Users.name == "lqz").all()
r4 = session.query(Users).filter_by(name='lqz').all()
r5 = session.query(Users).filter_by(name='lqz').first()
#:value 和:name 至关于占位符,用params传参数
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
#自定义查询sql
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()
"""

#增,删,改都要commit()
session.close()

8.经常使用操做


# 条件
ret = session.query(Users).filter_by(name='lqz').all()
#表达式,and条件链接
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
#注意下划线
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
#~非,除。。外
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
#二次筛选
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
#or_包裹的都是or条件,and_包裹的都是and条件
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
   or_(
       Users.id < 2,
       and_(Users.name == 'eric', Users.id > 3),
       Users.extra != ""
  )).all()


# 通配符,以e开头,不以e开头
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制,用于分页,区间
ret = session.query(Users)[1:2]

# 排序,根据name降序排列(从大到小)
ret = session.query(Users).order_by(Users.name.desc()).all()
#第一个条件重复后,再按第二个条件升序排
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
#分组以后取最大id,id之和,最小id
ret = session.query(
   func.max(Users.id),
   func.sum(Users.id),
   func.min(Users.id)).group_by(Users.name).all()
#haviing筛选
ret = session.query(
   func.max(Users.id),
   func.sum(Users.id),
   func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表(默认用forinkey关联)

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
#join表,默认是inner join
ret = session.query(Person).join(Favor).all()
#isouter=True 外连,表示Person left join Favor,没有右链接,反过来便可
ret = session.query(Person).join(Favor, isouter=True).all()
#打印原生sql
aa=session.query(Person).join(Favor, isouter=True)
print(aa)
# 本身指定on条件(连表条件),第二个参数,支持on多个条件,用and_,同上
ret = session.query(Person).join(Favor,Person.id==Favor.id, isouter=True).all()
# 组合(了解)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()

9.执行原生sql


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 db import Users, Hosts

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

# 查询
# cursor = session.execute('select * from users')
# result = cursor.fetchall()

# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'lqz'})
session.commit()
print(cursor.lastrowid)

session.close()

10.一对多


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 db import Users, Hosts, Hobby, Person

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
  Hobby(caption='乒乓球'),
  Hobby(caption='羽毛球'),
  Person(name='张三', hobby_id=3),
  Person(name='李四', hobby_id=4),
])

person = Person(name='张九', hobby=Hobby(caption='姑娘'))
session.add(person)
#添加二
hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飞'), Person(name='博雅')]
session.add(hb)

session.commit()
"""

# 使用relationship正向查询
"""
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)
"""

# 使用relationship反向查询
"""
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)
"""
#方式一,本身链表
# person_list=session.query(models.Person.name,models.Hobby.caption).join(models.Hobby,isouter=True).all()
person_list=session.query(models.Person,models.Hobby).join(models.Hobby,isouter=True).all()
for row in person_list:
   # print(row.name,row.caption)
   print(row[0].name,row[1].caption)

#方式二:经过relationship

person_list=session.query(models.Person).all()
for row in person_list:
   print(row.name,row.hobby.caption)
#查询喜欢姑娘的全部人
obj=session.query(models.Hobby).filter(models.Hobby.id==1).first()
persons=obj.pers
print(persons)
session.close()

11.多对多


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 db import Users, Hosts, Hobby, Person, Group, Server, Server2Group

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
  Server(hostname='c1.com'),
  Server(hostname='c2.com'),
  Group(name='A组'),
  Group(name='B组'),
])
session.commit()

s2g = Server2Group(server_id=1, group_id=1)
session.add(s2g)
session.commit()


gp = Group(name='C组')
gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
session.add(gp)
session.commit()


ser = Server(hostname='c6.com')
ser.groups = [Group(name='F组'),Group(name='G组')]
session.add(ser)
session.commit()
"""


# 使用relationship正向查询
"""
v = session.query(Group).first()
print(v.name)
print(v.servers)
"""

# 使用relationship反向查询
"""
v = session.query(Server).first()
print(v.hostname)
print(v.groups)
"""


session.close()

12.其它


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, func
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 关联子查询:correlate(Group)表示跟Group表作关联,as_scalar至关于对该sql加括号,用于放在后面当子查询
subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
result = session.query(Group.name, subqry)
"""
SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid
FROM server
WHERE server.id = `group`.id) AS anon_1
FROM `group`
"""
'''

select * from tb where id in [select id from xxx];

select id,
name,
#必须保证这次查询只有一个值
(select max(id) from xxx) as mid
from tb

例如,第三个字段只能有一个值
id name mid
1 lqz   1,2 不合理
2 egon   2


'''


# 原生SQL
"""
# 查询
cursor = session.execute('select * from users')
result = cursor.fetchall()

# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
session.commit()
print(cursor.lastrowid)
"""

session.close()

13.Flask-SQLAlchemy

flask和SQLAchemy的管理者,经过他把他们作链接


db = SQLAlchemy()
- 包含配置
- 包含ORM基类
- 包含create_all
- engine
- 建立链接

离线脚本,建立表

详见代码