Flask project construction 2

---

Preface

Tips ： Here you can add the general content to be recorded in this article ：

---

One 、 Directory framework

1.settings Catalog

Used to store configuration information
There are usually secret keys and SQLAlchemy Configuration information
For example, define two py file index dev Store separately

index.py

class Defaut():
    SECRET_KEY = 'hweyuwqq53461jeihjebjd34jf3'

dev.py

Inherit the previous configuration class Then configure the database

from settibgs.index import Defaut

class Dev(Defaut):
    SQLALCHEMY_DATABASE_URI = 'mysql://root:[email protected]:3306/sjk527'
    SQLALCHEMY_ECHO = False
    SQLALCHEMY_TRACK_MODIFICATIONS = True

2.create_flask.py

Used to define factory functions Generate application objects app Get configuration information
Registered blueprint Initialize project Hook method Configure cross domain

from flask import Flask
from flask_cors import CORS
from common.utils.jwt_auth import jwt_authentication
from models import db
from views.user import user_bp

def create_flask_app(info):
    #  Create an app
    app = Flask(__name__)
    #  Get configuration information from the object 
    app.config.from_object(info)
    #  Registered blueprint  
    app.register_blueprint(user_bp)
    #  Initialize the entire project 
    db.init_app(app)
    #  Configure cross domain 
    CORS(app)
    #  Use hook method   verification token
    app.before_request(jwt_authentication)
    return app

3.main.py

Start the main entrance of the project
You need to import the previous configuration file and factory function

from create_flask import create_flask_app
from settibgs.dev import Dev

app = create_flask_app(Dev)

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

4.models Catalog !

Used to store model classes
You need to create a __init__.py To create a SQLAlchemy object
It is generally named db

init.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

users.py

Here is an example of users and roles One-to-many relation

#  One to many 
    class User(db.Model):
        __tablename__ = 'tb_user'
        uid = db.Column(db.Integer, doc=' user id')
        name = db.Column(db.String(33), doc=' user name ')
        phone = db.Column(db.String(11), doc=' cell-phone number ')
        role_id = db.Column(db.Integer, db.ForeignKey('tb_role.rid'), doc=' Role foreign key ')

    class Role(db.Model):
        __tablename__ = 'tb_role'
        rid = db.Column(db.Integer, doc=' role id')
        name = db.Column(db.String(33), doc=' The role of ')

        r_user = db.relationship('User', backref='roles')

6.migrate.py Migration scripts

Generate Manager object
transfer
Add migration command
Migrate the main entrance
start-up Manager
The migration script must import all models , Otherwise, the required table will not be generated

from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager

from main import app
from models import db
from models.user import *

manage = Manager(app)
migrate = Migrate(app, db) #  here db Represents the migration of all tables 
manage.add_command('db', MigrateCommand) # 'db' Migrate keywords 

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

7.model_fields Catalog

Used to store the fields that need to be serialized
Import required fields
collocation marshal Use
for example : The following integer character string Time Boolean type

from flask_restful import fields
#  Article fields 
news_fields = {
    
    'nid': fields.Integer,
    'title': fields.String,
    'ctime': fields.DateTime,
    'is_follow':fields.Boolean,
}

8 views View Directory

Used to store the required view files
Blueprints are generally used
collocation falsk_restful
for example : Send SMS verification code view

An error was reported here

1) Before different blueprints url Cannot be the same url_prefix=' Cannot be the same '

import json
import random
import redis
from flask import Blueprint, jsonify
from flask_restful import Api, Resource, reqparse
from werkzeug.security import generate_password_hash
from common.celery.tasks import sendSmsCode
from common.utils.jwtToken_generate_check import generate_token
from models.user import *

#  Generate blueprints  
user_bp = Blueprint('user_bp', __name__, url_prefix='/v1_0')
#  Generate api object 
api = Api(user_bp)

#  Send SMS verification code   Configure Ronglian cloud  celery asynchronous   Generate   send out 
class SendSmsCode(Resource):  #  Inherit Resource

    def post(self):
        smsCode = f'{
      random.randint(100000, 999999)}'
        print(' The generated SMS verification code is ====={', smsCode)
        parser = reqparse.RequestParser()
        parser.add_argument('mobile')
        args = parser.parse_args()
        mobile = args['mobile']
        print(' The mobile number from the front end ', mobile)
        #
        ret = sendSmsCode.delay(mobile, smsCode)
        if ret:   #  Send successfully   Deposit in redis
            rds = redis.Redis(db=2)
            key = f'mobile:{
      mobile}:smsCode'
            rds.set(key, smsCode, ex=2*60*60)
            rds.close()
            return jsonify(msg=' The mobile number was sent successfully ,  Pay attention to check ', code=200)
        else:
            return jsonify(msg=' Mobile number sending failed ', code=500)

#  Send SMS verification code 
api.add_resource(SendSmsCode, '/sms_code')

9 common Catalog

General catalog Current storage :
celery Asynchronous texting #
Generate token check token #
Authentication hook #
Force login #
Timing task
Qiniuyun
cache

An error was reported here

1 Rong Lianyun sends SMS verification code
1)@app.task Use it above a function to be asynchronous delay
2) Turn on celery Asynchronous commands : File directory tasks
celery -A The file directory of the task to be performed worker -l info -P eventlet
2 Generate tokrn
1) Generate token Needed salt current_app Two call modes
secret = current_app.config.get(‘SECRET_KEY’)
secret = current_app.config[‘SECRET_KEY’]

3 Force login

1)`@wraps(func) `

 Force login   Define decorators 
def login_qlvi(func):
    @wraps(func)
    def wrap(*args, **kwargs):
        if g.user_id:
            return func(*args, **kwargs)
        else:
            return {
    'msg': ' Please log in first '}, 401
    return wrap

Generate token check token

Generate token And check token Be careful s The difference between
algorithm And algorithms

import datetime
import jwt
from flask import current_app

# TODO  An error log   Get configuration data from application context  current_app.config[' name ']
#  Package generation token
def generate_token(payload, exprie):
    secret = current_app.config.get('SECRET_KEY')
    payload.update({
    'exp': exprie})
    token = jwt.encode(payload, secret, algorithm='HS256')
    print(' This is the first generation token', token)
    return token

#  check token
def check_token(token, algorithms='HS256'):
    secret = current_app.config.get('SECRET_KEY')
    try:
        payload = jwt.decode(token, secret, algorithm)
        return payload
    except:
        return None


#  Generate two token
def generate_token2(payload, is_refresh=True):
    exprie = datetime.datetime.now() + datetime.timedelta(hours=24)
    token = generate_token(payload, exprie)
    if is_refresh:
        exprie_ = datetime.datetime.now() + datetime.timedelta(days=15)
        payload.update({
    'is_refresh': True})
        refresh_token = generate_token(payload, exprie_)
    else:
        refresh_token = None

    return token, refresh_token

Authentication hook

from flask import g, request
from common.utils.jwtToken_generate_check import check_token

def token_authcxx():
    print('######## Enter the authentication hook ########')
    token = request.headers.get('Authorization')
    print(' Whether to obtain the front end in authentication token:===?', token)
    payload = check_token(token)
    if payload:
        g.user_id = payload['user_id']
    else:    #  Assign value when there is load g object   If not, let it go 
        pass

celery Asynchronous texting

celery_main.py To configure celery

from celery import Celery

app = Celery('myWorker', broker='redis://127.0.0.1:6379/5',
             backend='redis://127.0.0.1:6379/6')

app.autodiscover_tasks(['common.celery.tasks'])

tasks.py Task file

import json
from ronglian_sms_sdk import SmsSDK
from common.celery.celery_main import app

accId = '8a216da8804bxa8a5018064a39933069b'
accToken = '09880a5e3fc94046b3cc1x47e240ad169'
appId = '8a216da8x804ba8a5018064a39a3706a2'

@app.task  #  If you don't write here, you can't delay Send asynchronously 
def sendSmsCode(mobile1, smsCode, expire=60):

    sdk = SmsSDK(accId, accToken, appId)
    tid = '1'
    mobile = mobile1
    datas = (f'{
      smsCode}', f'{
      expire}')

    res = sdk.sendMessage(tid, mobile, datas)
    print('res======>?', res)
    # return res
    if res:  #  Pass as long as there is a response   You can not write the real mobile phone number 
        return True
    else:
        return False
        
    #  This judgment is   You must send a successful text message to return Ture
    # if json.loads(res)['status Code'] == "000000":
    # return True
    # else:
    # return False

Generate seven cattle cloud token

from qiniu import Auth, put_file, etag
import qiniu.config
# You need to fill in your  Access Key  and  Secret Key
access_key = '74F2GCXwNJ2c9fgiD3ufHUUDXUrSdgNXW6kmIrXY'
secret_key = 'LqiDv9FZT8oLhDMHrLWwR-HP1vOoEguyQWUQlmNz'
def qiniu_token():    
    q = Auth(access_key, secret_key)  #  Build authentication object 
    bucket_name = 'p7lotk'  #  Space to upload 
    # Generate upload  Token, You can specify expiration time, etc 
    token = q.upload_token(bucket_name)
    return token

Call in view

class QiniuToken(Resource):
    @login_required
    def get(self):
        """ Generate seven cattle cloud upload token"""
        token = qiniu_token()
        print(' Seven cow cloud token======', token)
        return jsonify(msg=' Seven cattle cloud generated token', code=200, data={
    'token': token})

cache

User cache
Access to the cache Clear cache Determine if it's in the cache

import json
import random
import redis
from flask import jsonify
from redis import RedisError
from sqlalchemy.orm import load_only

from models.users import UserModel



#  Validity setting :
class BaseCacheTTL(object):

    TTL = 0  #  Set by subclass 
    MAX_DELTA = 10 * 60  #  Random increment upper limit 

    @classmethod
    def get_val(cls):
        return cls.TTL + random.randrange(0, cls.MAX_DELTA)


#  User profile data cache time 
class UserProfileCacheTTL(BaseCacheTTL):
    TTL = 30 * 60


# key Validity period when it does not exist 
class UserNotExistsCacheTTL(BaseCacheTTL):
    TTL = 1*60


#  Realization   cache 
class UserCacha():
    def __init__(self, user_id):
        self.rds = redis.Redis()
        self.user_id = user_id
        self.key = f'user:{
      user_id}:profile'

    #  Access to the cache 
    def get(self):
        """  Access to the cache  :return:  cache   Access to the cache  ---  There is   Return cache  --- non-existent   lookup mysql == Find the   Deposited in the cache   And return the query data  == Not found   non-existent   Set up Redis The value recorded in is -1 """
        # rds = redis.Redis()
        user_rds = self.rds.get(self.key)
        if user_rds:   #  There is   Return cache 
            user_rds1 = json.loads(user_rds)
            return user_rds1
        else:          #  non-existent   lookup mysql
            try:     # options(load_only( The model name . Field name ))  Lazy query   The query will be executed only when necessary 
                user = UserModel.query.filter_by(uid=self.user_id).first()
                print(' Query database user data :', user)
            except Exception as e:
                print(e)
                return jsonify(msg=' Database query operation failed ', code=500)

            #  Empty data found , redis Record as -1
            if user is None:
                self.rds.setex(self.key, UserNotExistsCacheTTL.get_val(), -1)
                # user_rds = rds.get(self.key)
                return None

            #  Find specific data   Return the data that needs to be cached 
            else:
                user_dict = {
    
                    'uname': user.uname,
                    'mobile': user.mobile,
                    'account': user.account,
                    'gender': user.gender,
                    'profile_photo': user.profile_photo,
                    'is_media': user.is_media,
                    'introduction': user.introduction,
                    'certificate': user.certificate,
                    'article_count': user.article_count,
                    'following_count': user.following_count,
                    'fans_count': user.fans_count,
                    'like_count': user.like_count
                }
                #  Deposit in redis
                self.rds.set(self.key, json.dumps(user_dict), ex=UserProfileCacheTTL.get_val())
                return user_dict

    #  Clear cache 
    def clear(self):
        # rds = redis.Redis()
        try:
            self.rds.delete(self.key)
        except RedisError as e:
            print(f' The error message of deleting the cache is {
      e}')

    #  Determine whether the user cache is implemented 
    def user_exists(self):
        #  Inquire about redis 1 There is  -1False  Otherwise return to Ture
        #  non-existent   Query the database   Query to   Deposit in   return Ture  We can't find it   save -1  return  Flase
        try:
            user_rds = self.rds.get(self.key)
        except Exception as e:
            print(f' Inquire about redis Data error {
      e}')
            user_rds = None
        if user_rds:
            if user_rds == b'-1':
                return False
            else:
                return True
        else:
            user = UserModel.query.filter_by(uid=self.user_id).first()
            #  Empty data found , redis Record as -1
            if user is None:
                self.rds.setex(self.key, UserNotExistsCacheTTL.get_val(), -1)
                # user_rds = rds.get(self.key)
                return None
            else:
                user_dict = {
    
                    'uname': user.uname,
                    'mobile': user.mobile,
                    'account': user.account,
                    'gender': user.gender,
                    'profile_photo': user.profile_photo,
                    'is_media': user.is_media,
                    'introduction': user.introduction,
                    'certificate': user.certificate,
                    'article_count': user.article_count,
                    'following_count': user.following_count,
                    'fans_count': user.fans_count,
                    'like_count': user.like_count
                }
                self.rds.setex(self.key, UserProfileCacheTTL.get_val(), json.dumps(user_dict))
                return user_dict

Use of cache in view

class UserInfo(Resource):
    """ Get user information from cache   Encapsulated cache class  UserCacha()  You need to transfer parameters when using  id . Function name ()  Call the corresponding method  """
    def get(self, id):
        # user = UserModel.query.filter_by(uid=id).first()
        # UserCacha(id).clear() #  Clear cache is called here 
        
        user = UserCacha(id).get()  #  Call here to get cache 
        if user:
            user11 = marshal(user, user_fields)
            return user11

Summary of subsequent error reports