当前位置：网站首页>Common components of flask

Common components of flask

2022-07-04 07:40:00 【A-L-Kun】

List of articles

Flask Common components

One 、 flask_session

session Is based on cookie Realization , The key value pair stored in the server （ In the form of { Random string ：'uuid'}）, At the same time in the browser cookie It also corresponds to the same random string , Used to request again When to verify , The function of this component is to session Data is stored in the database

1、 Common configuration 

#  Yes  session  To protect 
SECRET_KEY = os.urandom(32)  #  Create key 
SESSION_USE_SIGNER = True  #  Whether signature protection 

PERMANENT_SESSION_LIFETIME = timedelta(minutes=30)  #  Set the time delay 
SESSION_REFRESH_EACH_REQUEST = True  #  Update every time you refresh session

SESSION_TYPE = "redis"  #  Set up redis Storage session data 
SESSION_REDIS = Redis("127.0.0.1")  #  Connect to database 
SESSION_PREFIX = "MyWebPrefix:"  #  Set up session Prefix , The default is session:

Be careful ：session Stored in is a dictionary , When modifying the inside of the dictionary （ Level second ） Element time , Data will not be updated

resolvent ：

We need to read some of the source code first 

class SecureCookieSession(CallbackDict, SessionMixin):
    #: When data is changed, this is set to ``True``. Only the session
    modified = False  #  Only when session When the data inside changes , This will be set to True

    #: When data is read or written, this is set to ``True``. Used by
    accessed = False

    def __init__(self, initial: t.Any = None) -> None:
        def on_update(self) -> None:
            self.modified = True  #  If the data is updated , It will be changed to True, It will be right later session Set to cookies in 
            """ if not (session.modified #  If session modify , It will be reset to cookies in  or ( session.permanent #  Set to cookie Inside  and app.config["SESSION_REFRESH_EACH_REQUEST"] #  If True, Refresh every time  )): return ... response.set_cookie(...) """
            self.accessed = True  #  If session read , Will session write in response in 
            """ if session.accessed: response.vary.add("Cookie") """
		#  take on_update() Pass to CallbackDict
        super().__init__(initial, on_update)

resolvent ：

  1. Method 1 

    session.modified = True  #  Manually make session Marked as session Is modified

  2. Method 2 

    app.config["SESSION_REFRESH_EACH_REQUEST"] = True  #  Update every refresh 
#  meanwhile , After successful login , Set up 
session.permanent = True  #  The default is  False, But if you use redis Words , You don't need to set up , His tacit view is that True

2、 Usage method

2.1 session_interface

adopt session_interface To set up 

from flask_session import RedisSessionInterface
from flask import Flask
from os import urandom
from redis import Redis

app = Flask(__name__)
app.serect_key = urandom(32)  #  Set up 32 Bit random key 
app.config["SESSION_USE_SIGNER"] = True  
#  adopt redis preservation session
app.session_interface = RedisSessionInterface(
    redis=Redis("127.0.0.1"),  #  Connect Redis database 
    key_prefix="flask_login",  #  Set the prefix of random string , The default is session:
)

Default session Set to 

from flask.sessions import SecureCookieSessionInterface
app.session_interface = SecureCookieSessionInterface()

After modification , You can use it directly , Its use method is the same as the original one

2.2 config

Set... Through the configuration file 

from flask_session import Session
from flask import Flask
from os import urandom
from redis import Redis

settings = {
    
    "SESSION_TYPE": "redis",  #  Use redis Database connection 
    "SESSION_REDIS": Redis("127.0.0.1"),  #  Connect redis database 
    "SECRET_KEY": os.urandom(32),  #  Create key 
    "SESSION_USE_SIGNER": True,  #  Whether signature protection 
    "SESSION_REFRESH_EACH_REQUEST"： True, #  Update the data every time you refresh 
}
app.config.from_mapping(settings)  #  Set configuration 
Session(app)  #  The settings are encapsulated inside app.session_interface Methods

Two 、 DBUtils

1、 introduction

When we want to operate the database , You can do this ：

import pymysql
from functools import wraps
SQL_CONFIG = {
    
    #  Configuration of database 
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "passwd": "qwe123",
    "db": "flask1",
}

def sql_outer(fun):
    """ Use decorators , It can operate the database conveniently """
    @wraps(fun)
    def inner(sql, *args):
        conn = pymysql.connect(**SQL_CONFIG)  #  Connect to database 
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)  #  Returns the dictionary type SQL data 
        obj = fun(sql, cursor, *args)  #  Pass parameters into the function , Run 
        conn.commit()  #  Commit transaction 
        cursor.close()  #  Close cursor 
        conn.close()  #  Close the connection 
        return obj  #  Return the data 

    return inner


class SQLTool:
    @staticmethod
    @sql_outer
    def fetch_all(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        # print(*args)
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject 
        obj = cursor.fetchall()  #  Get all the data 
        return obj

    @staticmethod
    @sql_outer
    def fetch_one(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject , At the same time, unpack tuples 
        obj = cursor.fetchone()  #  Get the first data 
        return obj

if __name__ == '__main__':
    obj = SQLTool.fetch_one("SELECT id, name FROM users WHERE name=%s and pwd=%s", ["kun", "123"])
    print(obj)

The problem is coming. , If there are many connections , Is it troublesome to open the database and then close it ？

  • We can use database connection pool ：DBUtils

2、 DBUtils

Use database connection pool

This connection pool has two modes ：

  1. Create a connection for each thread , Even if the thread calls close Method , It won't shut down , Just resend the connection to the connection pool , For its own thread to use again . When the thread terminates , Connection automatically closed
  2. Create a batch of connections to the connection pool , For all threads to share （ The main ）

2.1 Model a 

POOL = PersistentDB(
    creator=pymysql,  #  Use the linked database module 
    maxusage=None,  #  The maximum number of times a link is reused ,None Means unlimited 
    setsession=[],  #  List of commands to execute before starting a session . Such as ：["set datestyle to ...", "set time zone ..."]
    ping=0,
    # ping MySQL Server side , Check if the service is available .#  Such as ：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
    closeable=False,
    #  If False when , conn.close()  Actually ignored , For next use , When the thread is closed again , Will automatically close the link . If True when , conn.close() Then close the link , Then call... Again pool.connection It's a mistake , Because the connection is really closed （pool.steady_connection() You can get a new link ）
    threadlocal=None,  #  This thread only enjoys the worthy object , Used to save linked objects , If the linked object is reset 
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='pooldb',
    charset='utf8'
)

def func():
    conn = POOL.connection(shareable=False)
    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    cursor.close()
    conn.close()

func()

2.2 Model 2 

import time
import pymysql
import threading
from dbutils.pooled_db import PooledDB, SharedDBConnection
POOL = PooledDB(
    creator=pymysql,  #  Use the linked database module 
    maxconnections=6,  #  The maximum number of connections allowed by the connection pool ,0 and None Indicates that there is no limit to the number of connections 
    mincached=2,  #  On initialization , At least free links created in the link pool ,0 Means not to create 
    maxcached=5,  #  The most idle links in the link pool ,0 and None Don't limit 
    maxshared=3,  #  The maximum number of links shared in the link pool ,0 and None Means share all .PS:  It's useless , because pymysql and MySQLdb Wait for the module  threadsafety All for 1, All values regardless of setting to ,_maxcached For ever 0, So always all links are shared .
    blocking=True,  #  If there is no connection available in the connection pool , Whether to block waiting .True, wait for ;False, Don't wait and report an error 
    maxusage=None,  #  The maximum number of times a link is reused ,None Means unlimited 
    setsession=[],  #  List of commands to execute before starting a session . Such as ：["set datestyle to ...", "set time zone ..."]
    ping=0,
    # ping MySQL Server side , Check if the service is available .#  Such as ：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='123',
    database='pooldb',
    charset='utf8'
)


def func():
    #  Check if the number of currently running connections is less than the maximum number of links , If not less than then ： Waiting or reporting raise TooManyConnections abnormal 
    #  otherwise 
    #  The priority is to get the link from the link created during initialization  SteadyDBConnection.
    #  And then SteadyDBConnection Objects are encapsulated in PooledDedicatedDBConnection And in return .
    #  If the link you first created doesn't have a link , To create a SteadyDBConnection object , Repackaged to PooledDedicatedDBConnection And in return .
    #  Once the link is closed , The connection is returned to the connection pool for subsequent threads to continue using .
    conn = POOL.connection()

    # print(th, ' The link was taken away ', conn1._con)
    # print(th, ' There are... In the pool at present ', pool._idle_cache, '\r\n')

    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    conn.close()

func()

3、 Code encapsulation 

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : sql.py
# @time : 2022/6/5 0:29
import pymysql
from functools import wraps
#  Create a database connection pool using singleton mode 
from dbutils.pooled_db import PooledDB
import pymysql
SQL_CONFIG = {
    
    #  Configuration of database 
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "passwd": "qwe123",
    "db": "flask1",
    "charset": "utf8",
}
POOL_CONFIG = {
    
    #  Configuration of database connection pool 
    "maxconnections": 6,
    "mincached": 2,
    "maxcached": 5,
    "maxshared": 3,
    "blocking": True,
    "maxusage": None,
    "setsession": [],
    "ping": 0,
}

POOL = PooledDB(
    creator=pymysql,
    **SQL_CONFIG,
    **POOL_CONFIG
)

def select_sql(type_):
    def sql_outer(fun):
        """ Use decorators , It can operate the database conveniently """

        @wraps(fun)
        def inner(sql, *args):
            if type_ == "sql":
                """ If the value is obtained through the database """
                conn = pymysql.connect(**SQL_CONFIG)  #  Connect to database 
                cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)  #  Returns the dictionary type SQL data 
                obj = fun(sql, cursor, *args)  #  Pass parameters into the function , Run 
                conn.commit()  #  Commit transaction 
                cursor.close()  #  Close cursor 
                conn.close()  #  Close the connection 
                return obj  #  Return the data 
            elif type_ == "pool":
                #  Take values through the database connection pool 
                conn = POOL.connection()
                cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
                obj = fun(sql, cursor, *args)
                conn.commit()
                cursor.close()  #  Close cursor 
                conn.close()  #  Close the connection 
                return obj
            raise ValueError("type_ value error, value = pool or sql")
        return inner

    return sql_outer


class SQLTool:
    """ Use pymysql Single threaded connection """

    @staticmethod
    @select_sql("sql")
    def fetch_all(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        # print(*args)
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject 
        obj = cursor.fetchall()  #  Get all the data 
        return obj

    @staticmethod
    @select_sql("sql")
    def fetch_one(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject , At the same time, unpack tuples 
        obj = cursor.fetchone()  #  Get the first data 
        return obj


class DBUtilsTool:
    """ Use database connection pool """

    @staticmethod
    @select_sql("pool")
    def fetch_all(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        # print(*args)
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject 
        obj = cursor.fetchall()  #  Get all the data 
        return obj

    @staticmethod
    @select_sql("pool")
    def fetch_one(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject , At the same time, unpack tuples 
        obj = cursor.fetchone()  #  Get the first data 
        return obj

obj = DBUtilsTool.fetch_one("SELECT id, name FROM users WHERE name=%s and pwd=%s", ["kun", "123"])
print(obj)

4、 combination flask Use

4.1 Mode one

The reason why it cannot be created is flask Context management mechanism , When flask Not yet run() When , cannot access current_app Configuration information inside

Put the configuration information into settings In profile , Reuse init_app Method —— simulation Session(app), Put the database connection pool into the configuration file config Inside

In the file where the database connection pool is stored sql.py in ：

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : sql.py
# @time : 2022/6/5 0:29
import pymysql
from functools import wraps
from flask import current_app
from dbutils.pooled_db import PooledDB


def sql_outer(fun):
    """ Use decorators , It can operate the database conveniently """

    @wraps(fun)
    def inner(sql, *args):
        #  Take values through the database connection pool 
        POOL = current_app.config["SQL_POOL"]  #  Store it in the configuration file , Prevent the connection pool from not existing 
        conn = POOL.connection()
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        obj = fun(sql, cursor, *args)
        conn.commit()
        cursor.close()  #  Close cursor 
        conn.close()  #  Close the connection 
        return obj

    return inner


class DBUtilsTool:
    """ Use database connection pool , simulation Session(app)"""
    def __init__(self, app=None):
        self.app = app
        if app:
            self.init_app(app)

    def init_app(self, app):
        POOL = PooledDB(
            **app.config["POOL_CONFIG"]
        )
        app.config["SQL_POOL"] = POOL

    @staticmethod
    @sql_outer
    def fetch_all(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        # print(*args)
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject 
        obj = cursor.fetchall()  #  Get all the data 
        return obj

    @staticmethod
    @sql_outer
    def fetch_one(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject , At the same time, unpack tuples 
        obj = cursor.fetchone()  #  Get the first data 
        return obj

stay settings.py In file ：

import pymysql

class BaseConfig:
    POOL_CONFIG = {
    
        #  Configuration of database connection pool 
        "creator": pymysql,
        "maxconnections": 6,
        "mincached": 2,
        "maxcached": 5,
        "maxshared": 3,
        "blocking": True,
        "maxusage": None,
        "setsession": [],
        "ping": 4,
        #  Configuration of database 
        "host": "127.0.0.1",
        "port": 3306,
        "user": "root",
        "passwd": "qwe123",
        "db": "flask1",
        "charset": "utf8",
    }

stay manage.py In the main running program 

from sql import DBUtilsTool
from flask import Flask

def create_app():
    app = Flask(__name__)
    app.config.from_object("settings.BaseConfig")
    #  simulation session_redis  take app Pass in , Initialize database connection pool 
    DBUtilsTool(app)  #  At the beginning of the program ,flask It hasn't started yet , Just put the configuration file in it ,flask When it's on , The database connection pool has been stored in config in 
    return app


if __name__ == '__main__':
    app = create_app()
    app.run("0.0.0.0")

4.2 Mode two

Add the database connection pool directly to the configuration file

stay sql.py in 

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : sql.py
# @time : 2022/6/5 0:29
import pymysql
from functools import wraps
from flask import current_app


def sql_outer(fun):
    """ Use decorators , It can operate the database conveniently """

    @wraps(fun)
    def inner(sql, *args):
        #  Take values through the database connection pool 
        POOL = current_app.config["SQL_POOL"]  #  Store it in the configuration file , Prevent the connection pool from not existing 
        conn = POOL.connection()
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        obj = fun(sql, cursor, *args)
        conn.commit()
        cursor.close()  #  Close cursor 
        conn.close()  #  Close the connection 
        return obj

    return inner


class DBUtilsTool:
    """ Use database connection pool """
    @staticmethod
    @sql_outer
    def fetch_all(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        # print(*args)
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject 
        obj = cursor.fetchall()  #  Get all the data 
        return obj

    @staticmethod
    @sql_outer
    def fetch_one(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject , At the same time, unpack tuples 
        obj = cursor.fetchone()  #  Get the first data 
        return obj

stay settings.py in 

from dbutils.pooled_db import PooledDB

class BaseConfig:
    SQL_POOL = PooledDB(
                #  Configuration of database connection pool 
                "creator": pymysql,
                "maxconnections": 6,
                "mincached": 2,
                "maxcached": 5,
                "maxshared": 3,
                "blocking": True,
                "maxusage": None,
                "setsession": [],
                "ping": 4,
                #  Configuration of database 
                "host": "127.0.0.1",
                "port": 3306,
                "user": "root",
                "passwd": "qwe123",
                "db": "flask1",
                "charset": "utf8",
            )

stay manage.py in 

from sql import DBUtilsTool
from flask import Flask

def create_app():
    app = Flask(__name__)
    app.config.from_object("settings.BaseConfig")
    return app


if __name__ == '__main__':
    app = create_app()
    app.run("0.0.0.0")

Method 2 is not recommended , It does not achieve the effect of separating configuration from program

4.3 Method 3

Use pool.py Create inside POOL

from flask import current_app


POOL = PooledDB(
                **current_app.config["POOL_CONFIG"]
            )

stay sql.py in 

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : sql.py
# @time : 2022/6/5 0:29
import pymysql
from functools import wraps
# from .pool import POOL #  Note that you cannot import here , because app Creation time , When registering blueprints ,app Not running , so current_app There is no configuration information in it 

def sql_outer(fun):
    """ Use decorators , It can operate the database conveniently """

    @wraps(fun)
    def inner(sql, *args):
        #  Take values through the database connection pool 
        from .pool import POOL  #  But when calling in the route, import ,app It must be running , so current_app There must be configuration information in it 
        conn = POOL.connection()
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        obj = fun(sql, cursor, *args)
        conn.commit()
        cursor.close()  #  Close cursor 
        conn.close()  #  Close the connection 
        return obj

    return inner


class DBUtilsTool:
    """ Use database connection pool """
    @staticmethod
    @sql_outer
    def fetch_all(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        # print(*args)
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject 
        obj = cursor.fetchall()  #  Get all the data 
        return obj

    @staticmethod
    @sql_outer
    def fetch_one(sql: "sql  sentence ", cursor=None, *args: " Data to query "):
        cursor.execute(sql, *args)  #  Note that string splicing cannot be used , prevent SQL Inject , At the same time, unpack tuples 
        obj = cursor.fetchone()  #  Get the first data 
        return obj

it is to be noted that current_app The founding period of , Other methods of use are the same

Just write native SQL, Use the database connection pool

3、 ... and 、 wtforms

effect ： Used exclusively for python web The framework does form validation

1、 Supported fields

Field describe
StringField The text field
TextAreaField Multiline text fields
PasswordField Password text field
HiddenField Hide text fields
DateField The text field , The value is datetime.date Format
DateTimeField The text field , The value is datetime.datetime Format
IntegerField The text field , The value is integer.
DecimalField The text field , The value is decimal.Decimal
FloatField The text field , The value is a floating-point number
BooleanField Check box , The value is True and False
RadioField A set of menus
SelectField The drop-down list
SelectMultipleField The drop-down list , Multiple values can be selected
FileField File upload field
SubmitField Form submit button
FormField Embed another form field
FieldList A set of fields of a specified type

2、 Field parameters

Parameters describe
label Field alias , In the page can be through the field .label Exhibition
validator List of validation rules
filters List of chlorine perchlorators , Used to filter the submitted data
description Description information , Usually used to generate help information
id It means that form The position of the field when the class is defined , Usually you don't have to define it , The default is to sort in the order defined
default The default value is
widgethtml plug-in unit , With this plug-in you can override the default plug-in , More through user customization
render_kw Customize html attribute
choices Check the options of the type

3、 Validation function

Validation function explain
Email Verification is an email address
EqualTo Compare the values of two fields ; It is often used in the case that you need to enter the key twice for confirmation
IPAddress verification IPv4 network address
Length The length of the input string
NumberRange Verify that the value entered is in the numeric range
Optional Skip other validation functions when there is no input value
DataRequired Make sure there is data in the field
Regexp Use regular expressions to validate input values
URL verification url
AnyOf Make sure the input value is in the list of optional values
NoneOf Make sure the input value is not in the optional list

According to the source code of the verification function , Customize the validation function

4、 Code instance

Create a function of uploading files app, It contains the analysis of the key steps of its workflow , Hope it is useful for understanding

Code instance ：

stay app.py in 

from flask import Flask
from wtforms import FileField, Form, SubmitField, widgets, validators
from flask import (request, render_template, current_app)
from werkzeug.utils import secure_filename
import os


class UploadForm(Form):
    """ Create fields for uploading files """
    # UploadForm.checkbox = UnboundField(BooleanField, *args, **kwargs, creation_counter=1)
    file1 = FileField(
        render_kw={
    "class": "file"},
        widget=widgets.FileInput(),
        validators=[
            validators.DataRequired(message=" Please select the file to upload ！")
        ]
    )
    # UploadForm.sbtn = UnboundField(SubmitField, *args, **kwargs, creation_counter=2)
    sbtn = SubmitField(
        render_kw={
    
            "id": "upload_submit",
            "value": " Upload files ",
        },
        widget=widgets.SubmitInput(),
    )


config = {
    
    "UPLOAD_FOLDER": "upload/",  #  Set the folder where files are stored 
}
app = Flask(__name__)
app.config.from_mapping(config)


@app.route("/upload", methods=['GET', 'POST'])
def upload():
    #  The route used to specially handle the submitted data 
    if request.method == "POST":
        form = UploadForm(formdata=request.files)  #  Upload files 
        if form.validate():  #  Validate the data 
            f = form.file1.data
            f.save(os.path.join(current_app.config['UPLOAD_FOLDER'], secure_filename(f.filename)))  #  Handle the file name safely 
            return render_template("index.html", form=form, msg=f" Upload file successful ！ The uploaded file is {
      f.filename}")
        else:
            return render_template("index.html", form=form, msg=" Please select a file ！")
    form = UploadForm()
    """ #  Analyze the source code when creating this class  metaclass=FormMeta #  It is a metaclass  #  so , When you create an object  1、 FormMeta.__call__ #  stay  __call__  Method  UploadForm._unbound_fields = None UploadForm._wtforms_meta = None _unbound_fields = [ #  Its basis counter Sort by  ("checkbox": UnboundField(BooleanField, *args, **kwargs, creation_counter=1)) ("sbtn": UnboundField(SubmitField, *args, **kwargs, creation_counter=1)) ] _wtforms_meta = type("Meta", tuple([DefaultMeta]), {}) # DefaultMeta = Form.Meta = class Meta(DefaultMeta): pass 2、 UploadForm.__new__ #  stay  __new__  Method , And then it turns out that there's no  __new__  Method , Unless you customize  pass 3、 UploadForm.__init__ #  perform  __init__  Method  UploadForm()._fields = { "file1": FileField(...), "sbtn": SubmitField(...) } UploadForm().name=FileField(...) UploadForm().sbtn=SubmitField(...) """
    print(form.file1)
    """ #  Source code analysis when accessing class properties  #  so ,form.name  It executes  __str__  Method  Field.__str__ -> return self() #  Execution field  __call__  Method  Field.__call__ -> return self.meta.render_field(self, kwargs) DefaultMeta.render_field(self, kwargs) -> return field.widget(field, **render_kw) #  perform  widget  Of  __call__  Method  Input.__call__ -> return Markup("<input %s>" % self.html_params(name=field.name, **kwargs)) #  Rendering  """
    for i in form:
        print(i)  #  Find out i It can be traversed ,
        """ #  There is a iter Method  BaseField.__iter__ -> iter(self._fields.values()) print(i) #  Execute  __str__  Method  """
    return render_template("index.html", form=form)


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

And the verification process , Try it on your own

stay templates/index.html in 

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title> Upload files </title>
</head>
<body>
    <div class="uploadFile">
        <h1> Please select the file to upload </h1>
        <form method="POST" enctype="multipart/form-data">
            <p>{
    {
     form.file1 }}</p>
            <p style="color: red">{
    {
     form.file1.errors[0] }}</p>
            <p>{
    {
     form.sbtn }}</p>
        </form>
        <p style="color: green">{
    {
     msg }}</p>
    </div>
</body>
</html>

Four 、 flask_mail

1、 brief introduction

stay web In the program , You often need to send email . such as , Send a confirmation email when the user registers an account ; Send popular content or promotional information to users regularly . stay Web Sending email in the program is not complicated , With expansion Flask-Mail Or a third-party email service , It only takes a few lines of code to send an email

Configuration information ：

Parameters describe
MAIL_SERVER The name of the mail server /IP Address
MAIL_PORT Port number of the server used
MAIL_USE_TLS Enable / Disable transport security layer encryption
MAIL_USE_SSL Enable / Disable secure socket layer encryption
MAIL_DEBUG Debugging support , The default is Flask Debug state of the application
MAIL_USERNAME The user name of the sender
MAIL_PASSWORD The sender's password
MAIL_DEFAULT_SENDER Set the default sender
MAIL_MAX_EMAILS Set the maximum messages to send
MAIL_SUPPRESS_SEND If app.testing Set to true, Then the transmission is suppressed
MAIL_ASCII_ATTACHMENTS If set to true, Convert the attached file name to ASCII

2、 Mail

It manages the requirements for email messages . Class constructors take the form

Method describe
send() send out Message The content of the class object
connect() Open connection with mail host
send_message() Send message to

3、 Massage

3.1 Instantiate objects 

Message(
    subject='',  #  Set title 
    recipients=[],  #  The recipient 
    body=None,  #  Content 
    html=None,
    sender=None,  #  Sender 
    cc=None,
    bcc=None,
    attachments=None,  #  The attachment 
    reply_to=None,  
    date=None,
    charset=None,
    extra_headers=None,
    mail_options=None,
    rcpt_options=None
)

3.2 Class method

attach() - Add attachments to the message . The method uses the following parameters ：

  • filename - The name of the file to attach
  • content_type - Of documents MIME type
  • data - Raw file data
  • disposition - Content handling , If any

add_recipient() - Add another recipient to the message

4、 Usage method

  1. Initialize mailbox 

    from flask_mail import Mail, Message
mail = Mail(app)

  2. Configure mailbox information 

    class BaseConfig2():
    MAIL_SERVER = "smtp.qq.com"  #  Set up SMTP The server 
    MAIL_PORT = 465  #  Set port 
    MAIL_USE_TLS = False  #  Whether to use TLSSL encryption 
    MAIL_USER_SSL = True  #  Whether to use SSL encryption 
    MAIL_USERNAME = "[email protected]"  #  mailbox 
    MAIL_PASSWORD = "xtisaddfdfntdcjf"  #  password 
    MAIL_DEFAULT_SENDER = "A.L.Kun<[email protected]>"  #  Sender

  3. Create information 

    msg = Message(subject="This is title", recipients="[email protected]")
msg.body = "This is body"

  4. Send a message 

    mail.send(msg)

Master code ：

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "app_.py"
__time__ = "2022/6/24 19:56"
__email__ = "[email protected]"
from settings import BaseConfig2
from flask import Flask
from flask_mail import Mail, Message
app = Flask(__name__)
app.config.from_object(BaseConfig2)
mail = Mail(app)


@app.route('/')
def hello():
    msg = Message(subject="This is title", recipients="[email protected]")
    msg.body = "This is body"
    message.html = render_template('content.html')  #  Can send a html page 
    mail.send(msg)
    return " Message sending completed ！"

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

5、 ... and 、 flask_script

1、 brief introduction

install ：pip install flask-script

By using Flask-Script Expand , We can do it in Flask When the server starts , Pass in the parameters... On the command line . Not just through app.run() In the method, pass on the parameter , For example, we can go through python hello.py runserver –host ip Address , Tell the server which network interface to listen for connections from clients

2、 Start the service 

from flask import Flask

app = Flask(__name__)

""" Use flask-script Start project """
from flask_script import Manager
manager = Manager(app)

@app.route('/')
def index():
    return 'hello world'

if  __name__ == "__main__"
    manager.run()

Enter at the command line python manage.py runserver, Start the service

3、 Pass in the parameter 

from flaskScript import create_app
from flask_script import Manager

app = create_app()  #  Return to one app object 
manager = Manager(app)


@manager.command
def custom(arg):
    """  This method can receive parameters input from the command line , Positional arguments   Such as ：  function ：python manage.py custom 123  It will print on the console  123 """
    print(arg)


@manager.option("-n", "--name", dest='name')
@manager.option("-u", "--url", dest="url")
def cmd_(name, url):
    """  Custom command , Optional parameters  :param name: Name passed in from the command line  :param url:  Passed in from the command line url :return:  Such as ：  Input ：python manage.py cmd_ -n "lihua" -u "127.0.0.1"  Will output ：lihua 127.0.0.1 """
    print(name, url)


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

effect ：

  • python manage.py runserver ...
  • python manage.py Custom command

6、 ... and 、 flask_sqlalchemy

1、 brief introduction

sqlalchemy There are sqlalchemy Some basic operations

flask Generally used in flask-sqlalchemy To manipulate the database , It's easy to use , Easy to operate

install ：

pip install flask-sqlalchemy

Configuration parameters

configuration option explain
SQLALCHEMY_DATABASE_URI Connect to database . Example ：mysql://username:[email protected]/post/db?charset=utf-8
SQLALCHEMY_BINDS A dictionary that will bind multiple databases . For more details, please see the official text Bind multiple databases .
SQLALCHEMY_ECHO Debug set to true
SQLALCHEMY_POOL_SIZE The size of the database pool , The default value is 5.
SQLALCHEMY_POOL_TIMEOUT Connection timeout
SQLALCHEMY_POOL_RECYCLE Number of seconds to automatically reclaim connections .
SQLALCHEMY_MAX_OVERFLOW The maximum number of connections that can be created is . When these extra After the connection is recycled to the connection pool, it will be disconnected and discarded .
SQLALCHEMY_TRACK_MODIFICATIONS If I set it to True ( By default ),Flask-SQLAlchemy Changes to the object will be tracked and signaled . This requires additional memory , You can disable it if you don't need it .

2、 Use steps

  1. Configuration database information , stay settings.py Add... To the file 

    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:[email protected]:3306/flask1?charset=utf8"

SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True

  2. Create a app.py file , Add... To it 

    from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()  #  Instantiate objects 
app = Flask(__name__)
app.config.from_file("settings.py")  #  Import profile 
db.init_app(app)  #  initialization sqlalchemy, Use app The configuration file inside

  3. Write the database structure , Create a models.py, And app.py At the same level 

    from app import db

class Users(db.Model):
    __table__ = "user"
    id = db.Column(db.INTEGER, primary_key=True, autoincrement=True)
    name = db.Column(db.String(32))

  4. Use in routing 

    import models

@app.route("/login", methods=["GET", "POST"])
def login():
    data = db.session.query(models.Users).all()  #  Find all the information in the database 
    print(data)
    db.session.remove()  #  remove session
    return "Login"

Expand , Use of offline scripts 

from app import app
with app.app_context():
    pass  #  It can run flask Operations performed at runtime

7、 ... and 、 flask_migrate

effect ： Do database migration

It depends on ：flask-script/flask-sqlalchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_script import Manager

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite:///app.db"  #  Connect to database 

db = SQLAlchemy(app)  
manager = Manager(app)
Migrate(app, db)  #  Instantiated component 

""" flask db init #  Initialize the table  flask db migrate #  Create the table in the database  flask db upgrade #  Update the structure of the table  """


# Model
class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))


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

8、 ... and 、 Custom components

stay auto.py Add ：

from flask import request, session, redirect


class Auth:
    def __init__(self, app=None):
        self.app = app
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        app.auto = self  #  Write all the information of this class app Inside 
        self.app = app
        self.app.before_request(self.check_login)

    def check_login(self):
        print(" Check whether the user is logged in ")
        usr = session.get("usr")
        print(usr)

    def login(self, data):
        """ establish session"""
        session["usr"] = data

    def login_out(self):
        """ The user to log out """
        del session["usr"]

When using this component ：

from auto import Auto
from flask import Flask


app = Flask(__name__)

at = Auto()
at.init_app(app)


################################################
from flask import current_app
cerrent_app.login_out()  #  Call the function of component login

Last , The general directory structure is ：

https://images.cnblogs.com/cnblogs_com/blogs/722174/galleries/2074790/o_220701084945_sort.png

Click on the I , View the source code

Nine 、 Other

1、 many app application 

from flask import Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware
from werkzeug.serving import run_simple

app01 = Flask("app01")
app02 = Flask("app02")
# //login ->  visit  app01 Below login
dm = DispatcherMiddleware(app01, {
    
    "/app02": app02  # /appo2/index  visit app02 Below index
})


@app01.route("/login")
def login():
    return "app01.login"


@app02.route("/index")
def index():
    return "app02.index"


if __name__ == '__main__':
    #  Once the request comes in, it will be executed  run_simple  The third parameter of the method is bracketed ：dm(), Adding parentheses to an object will call the __call__ Method 
    run_simple("127.0.0.1", 5000, dm)

2、 The signal 

from flask import Flask, signals

app = Flask(__name__)


def func(*args, **kwargs):
    print(" Request start , The trigger app by ", *args, **kwargs)


signals.request_started.connect(func)  #  Connecting signals 
"""  All signals contained  template_rendered before_render_template request_started request_finished request_tearing_down got_request_exception appcontext_tearing_down appcontext_pushed appcontext_popped message_flashed  Here you can check the trigger conditions of these signals in the source code , Source code internal use send Method trigger signal , Or Baidu OK """

@app.route("/")
def index():
    return " Homepage "


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

Be careful , The difference between a signal and a decorator is , The signal cannot control the progress of the program , It just provides a prompt function , Add additional operations and values to the original foundation ; The decorator can control whether the request can continue to be executed later

原网站

版权声明
本文为[A-L-Kun]所创，转载请带上原文链接，感谢
https://yzsam.com/2022/185/202207040735272401.html

边栏推荐

猜你喜欢

随机推荐