Flask Web開發從入門到放棄(一)

時間  2020-06-09
標籤 flask web 開發 入門 放棄 欄目 Python 简体版
原文   原文鏈接

 

  • 第1章 章節一

  • 01 內容概要

  • 02 內容回顧

  • 03 路飛學城之加入購物車

  • 04 路飛學城之結算

  • 05 路飛學城之當即支付

  • 06 路飛學城以後續計劃

  • 07 Flask框架簡介和快速使用

  • 08 FLask框架之用戶管理系統示例（一）

  • 09 Flask框架之用戶管理系統示例（二）

  • 第2章 章節二

  • 01 內容概要

  • 02 內容回顧

  • 03 Flask框架之配置

  • 04 Flask框架之建立路由的兩種方式

  • 05 Flask框架之反向生成URL

  • 06 Flask框架之自定義路由轉換器

  • 07 Flask框架之app.route參數

  • 08 Flask框架之獲取子域名的路由

  • 09 上述內容總結

  • 10 Flask框架之視圖中添加裝飾器

  • 11 Flask框架之CBV和FBV

  • 12 Flask框架之請求和響應相關

  • 13 Flask框架之模板引擎

  • 14 Flask框架之session使用和源碼流程（一）

  • 15 Flask框架之session使用和源碼流程（二）

  • 16 Flask框架之before_request和after_request(一)

  • 17 Flask框架之after_request(二)

  • 18 Flask框架字號常見裝飾器

  • 19 上述內容總結

  • 20 Flask之閃現

  • 21 Flask中間件

  • 22 Flask框架之藍圖（一）

  • 23 Flask框架之藍圖（二）

  • 24 拓展知識：pipreqs

  • 25 拓展知識：函數和方法

  • 第3章 章節三

  • 01 內容概要

  • 02 內容回顧

  • 03 threading.local學習

  • 04 自定義Local思路

  • 05 自定義Local對象（基於函數）

  • 06 自定義Local對象（基於面向對象）

  • 07 Flask上下文管理之本質分析

  • 08 Flask上下文管理之請求處處理階段

  • 09 Flask上下文管理之視圖調用階段

  • 10 Flask上下文管理之視圖調動階段和結束階段

  • 11 問題來了

  • 12 Flask中的g究竟是什麼呢？生命週期

  • 13 內容補充：面向對象的私有字段

  • 14 homework

  • 第4章 章節四

  • 01 內容概要

  • 02 內容回顧

  • 03 路飛學城補充：視頻播放受權

  • 04 flask-session組件使用和原理（一）

  • 05 flask-session組件使用和原理（二）

  • 06 flask-session組件使用和原理（三）

  • 07 flask-session組件使用和原理（四）

  • 08 基於pymysql實現用戶登陸

  • 09 基於數據庫鏈接池實現用戶登陸

  • 10 解決bug（一）

  • 11 解決bug（二）

  • 12 數據庫鏈接池總結

  • 13 WTforms介紹以及用戶登陸示例

  • 14 WTforms用戶註冊示例

  • 15 內容總結

  • 16 homework

第1章 章節一

01 內容概要

1.1 路飛購買流程

  • 加入購物車
  • 結算
  • 去支付

1.2 Flask框架

  • 路由
  • 視圖
  • 模板
  • session
  • ...

02 內容回顧

2.1 路飛學城項目結構

　　2.1.1 先後端分離；html

　　2.1.2 導師後臺+管理後臺+主站（本人負責）前端

2.2 主站的功能

　　2.2.1 Vue——兼職、課程列表、詳細、深科技

　　2.2.2 rest api

  • 課程系列——列表、詳細、推薦課程、章節&課時、常見問題、評論
  • 深科技——文章列表、詳細、評論、贊、收藏
  • 我的中心——個人帳戶、我的資料、訂單、課程中心
  • 購買流程（複雜）——加入購物車、去結算、當即支付
  • 其餘——關於咱們、聯繫咱們、意見反饋

2.3 技術點

　　2.3.1 rest framework框架——認證組件（用於用戶認證） or Django中間件，二者實現的時機不一樣，認證邏輯無需實現；

　　2.3.2 跨域——jsonp（動態生成script標籤） cors；

　　2.3.3 Redis——購物邏輯，用戶session兩個場景下使用；

  • 頻繁操做；
  • 中間狀態；
  • 數據放入內容，速度快；
  • Redis鎖

　　2.3.4 支付寶支付接口

  • RSA加密；
  • 數字金額有要求，保留小數點後兩位；
  • 兩個URL；
  • 支付寶公鑰和商戶私鑰；

　　2.3.5 微信消息推送

  • 微信企業號
  • 沙箱環境
  • 普通消息和模板消息
  • 關注公衆號，生成二維碼
  • 經過js生成二維碼
  • 惟一標識獲取到

　　2.3.6 ContenType

  • 參考Django settings
  • 參考Django 中間件

　　2.3.7 rest framework分頁

　　2.3.8 接口寫的最多的查詢接口；

　　2.3.9 視圖

  • queryset

　　2.3.10 序列化

  • __new__方法;
  • source
  • Method

　　2.3.11 Git協同開發

　　2.3.12 ORM操做

  • only
  • defer
  • exclude
  • filter

　　2.3.13 CRSF

  • 基於中間件作
  • 基於裝飾器

　　2.3.14 Vue.js基本命令

  • Vue.js的基本命令
  • Router攔截器
  • Ajax——jQuery、axios ——本質都是XMLHttpRequest對象實現；
  • 請求頭Content-Type：request.POST
  • json
  • vuex
  • vue-cookies

　　2.3.15 面試題總結準備

　　2.3.16 組織架構、人員配比、項目週期；

03 路飛學城之加入購物車

3.1 加入購物車 

#!/usr/bin/env python # -*- coding:utf-8 -*-
import json from django.core.exceptions import ObjectDoesNotExist from django.conf import settings from rest_framework.views import APIView from rest_framework.viewsets import ViewSetMixin from rest_framework.viewsets import ModelViewSet from rest_framework.response import Response from repository import models from api.serializer.payment import ShoppingCarSerializer from api.utils.auth.token_auth import LuffyTokenAuthentication from api.utils.auth.token_permission import LuffyPermission from api.utils import redis_pool from api.utils.exception import PricePolicyDoesNotExist class ShoppingCarView(ViewSetMixin, APIView): """ 購物車接口 """ authentication_classes = [LuffyTokenAuthentication, ] permission_classes = [LuffyPermission, ] def get(self, request, *args, **kwargs): """ 根據用戶ID獲取購物車全部東西 :param request: :param args: :param kwargs: :return: """ response = {'code': 1000, 'data': None} try: product_dict = redis_pool.conn.hget(settings.REDIS_SHOPPING_CAR_KEY, request.user.id) if product_dict: product_dict = json.loads(product_dict.decode('utf-8')) response['data'] = product_dict except Exception as e: response['code'] = 1001 response['msg'] = "獲取購物車列表失敗"

        return Response(response) def post(self, request, *args, **kwargs): """ # 根據課程ID獲取課程信息以及相關全部價格策略 chopping_car = { request.user.id:{ course.id:{ title:'xx', img:'xx', choice_policy_id:1, price_policy_dict:{ {id:1,price:'9.9', period:'1個月'}, {id:2,price:'19.9',period:'3個月'}, {id:3,price:'59.9',period:'8個月'}, }, } }， course.id:[ title:'xx', img:'xx', choice_policy_id:1, price_policy_dict:{ {id:1,price:'9.9', period:'1個月'}, {id:2,price:'19.9',period:'3個月'}, {id:3,price:'59.9',period:'8個月'}, }, ] } } } :param request: :param args: :param kwargs: :return: """ response = {'code': 1000, 'msg': None} try: course_id = int(request.data.get('course_id')) policy_id = int(request.data.get('policy_id')) # 獲取課程信息
            course = models.Course.objects.exclude(course_type=2).filter(status=0).get(id=course_id) # 序列化課程信息，並獲取其關聯的全部價格策略
            ser = ShoppingCarSerializer(instance=course, many=False) product = ser.data # 判斷價格策略是否存在
            policy_exist = False for policy in product['price_policy_list']: if policy['id'] == policy_id: policy_exist = True break
            if not policy_exist: raise PricePolicyDoesNotExist() # 設置默認選中的價格策略
            product.setdefault('choice_policy_id', policy_id) # 獲取當前用戶在購物車中已存在的課程，若是存在則更新，不然添加新課程
            product_dict = redis_pool.conn.hget(settings.REDIS_SHOPPING_CAR_KEY, request.user.id) if not product_dict: product_dict = {course_id: product} else: product_dict = json.loads(product_dict.decode('utf-8')) product_dict[course_id] = product # 將新課程寫入到購物車
 redis_pool.conn.hset(settings.REDIS_SHOPPING_CAR_KEY, request.user.id, json.dumps(product_dict)) except ObjectDoesNotExist as e: response['code'] = 1001 response['msg'] = '視頻不存在'
        except PricePolicyDoesNotExist as e: response['code'] = 1002 response['msg'] = '價格策略不存在'
        except Exception as e: print(e) response['code'] = 1003 response['msg'] = '添加購物車失敗'

        return Response(response) def delete(self, request, *args, **kwargs): """ 刪除購物車中的課程 :param request: :param args: :param kwargs: :return: """ response = {'code': 1000} try: course_id = kwargs.get('pk') product_dict = redis_pool.conn.hget(settings.REDIS_SHOPPING_CAR_KEY, request.user.id) if not product_dict: raise Exception('購物車中無課程') product_dict = json.loads(product_dict.decode('utf-8')) if course_id not in product_dict: raise Exception('購物車中無該商品') del product_dict[course_id] redis_pool.conn.hset(settings.REDIS_SHOPPING_CAR_KEY, request.user.id, json.dumps(product_dict)) except Exception as e: response['code'] = 1001 response['msg'] = str(e) return Response(response) def put(self, request, *args, **kwargs): """ 更新購物車中的課程的默認的價格策略 :param request: :param args: :param kwargs: :return: """ response = {'code': 1000} try: course_id = kwargs.get('pk') policy_id = request.data.get('policy_id') product_dict = redis_pool.conn.hget(settings.REDIS_SHOPPING_CAR_KEY, request.user.id) if not product_dict: raise Exception('購物車清單不存在') product_dict = json.loads(product_dict.decode('utf-8')) if course_id not in product_dict: raise Exception('購物車清單中商品不存在') policy_exist = False for policy in product_dict[course_id]['price_policy_list']: if policy['id'] == policy_id: policy_exist = True break
            if not policy_exist: raise PricePolicyDoesNotExist() product_dict[course_id]['choice_policy_id'] = policy_id redis_pool.conn.hset(settings.REDIS_SHOPPING_CAR_KEY, request.user.id, json.dumps(product_dict)) except PricePolicyDoesNotExist as e: response['code'] = 1001 response['msg'] = '價格策略不存在'
        except Exception as e: response['code'] = 1002 response['msg'] = str(e) return Response(response)

04 路飛學城之結算

4.1 結算 

#!/usr/bin/env python # -*- coding:utf-8 -*-
import json import datetime from django.conf import settings from rest_framework.views import APIView from rest_framework.response import Response from api.utils.auth.token_auth import LuffyTokenAuthentication from api.utils.auth.token_permission import LuffyPermission from api.utils import redis_pool from repository import models class PaymentView(APIView): """ 去結算接口 """ authentication_classes = [LuffyTokenAuthentication, ] permission_classes = [LuffyPermission, ] def get(self, request, *args, **kwargs): """ 獲取結算列表 :param request: :param args: :param kwargs: :return: """ response = {'code': 1000} try: # 結算商品列表
            payment_list = redis_pool.conn.hget(settings.REDIS_PAYMENT_KEY, request.user.id) if not payment_list: raise Exception() response['data'] = { 'payment_list': json.loads(payment_list.decode('utf-8')),  # 結算信息（課程、價格和優惠券）
                "balance": request.user.balance  # 我的貝里帳戶，可以使用貝里金額
 } except Exception as e: response['code'] = 1001 response['msg'] = "結算列表爲空"

        return Response(response) def post(self, request, *args, **kwargs): """ 去結算 方案一（示例）：用戶提交課程id，去redis購物車中獲取其選好的價格策略，再次檢測課程和價格策略的合法性。 PS: 直接購買時，須要先加入購物車，再當即去結算 方案二：用戶提交課程id和價格策略id，去數據庫驗證其合法性。 PS: 直接購買時，直接去結算 user.id: { policy_course_dict:{ 課程ID:{ 'course_id': course_id, 'course_name': product['name'], 'course_img': product['course_img'], 'policy_id': product['choice_policy_id'], 'policy_price': policy_price, 'policy_': policy_period, 'coupon_record_list': [ {'id': 0, 'text': '請選擇優惠券'}, {'id': 1, 'type':1, 'text': '優惠券1', ..}, {'id': 2, 'type':2, 'text': '優惠券1', ..}, {'id': 3, 'type':3, 'text': '優惠券1', ..}, ], }, 課程ID:{ 'course_id': course_id, 'course_name': product['name'], 'course_img': product['course_img'], 'policy_id': product['choice_policy_id'], 'policy_price': policy_price, 'policy_': policy_period, 'coupon_record_list': [ {'id': 0, 'text': '請選擇優惠券'}, {'id': 1, 'type':1, 'text': '優惠券1', ..}, {'id': 2, 'type':2, 'text': '優惠券1', ..}, {'id': 3, 'type':3, 'text': '優惠券1', ..}, ], } }, global_coupon_dict:{ 1:{'type': 0, 'text': "通用優惠券", 'id': 1, ..}, 2:{'type': 0, 'text': "通用優惠券", 'id': 2, ..}, 3:{'type': 0, 'text': "通用優惠券", 'id': 3, ...}, 4:{'type': 0, 'text': "通用優惠券", 'id': 4, ...}, } } :param request: :param args: :param kwargs: :return: """ response = {'code': 1001} try: """ 1. 獲取要支付的課程ID 2. 檢查購物車中是否存在，不存在則報錯 循環用戶提交的課程ID，去購物車中獲取，若是不存在，就報錯。 """
            # 獲取用戶提交的課程id
            course_id_list = request.data.get('course_list') if not course_id_list or not isinstance(course_id_list, list): raise Exception('請選擇要結算的課程') # 購物車中檢查是否已經有課程（應該有課程的）
            product_dict = redis_pool.conn.hget(settings.REDIS_SHOPPING_CAR_KEY, request.user.id) if not product_dict: raise Exception('購物車無課程') # 購物車中是否有用戶要購買的課程
            product_dict = json.loads(product_dict.decode('utf-8')) # ###### 課程、價格和優惠券 #######
            policy_course_dict = {} for course_id in course_id_list: course_id = str(course_id) product = product_dict.get(course_id) if not product: raise Exception('購買的課程必須先加入購物車') policy_exist = False for policy in product['price_policy_list']: if policy['id'] == product['choice_policy_id']: policy_price = policy['price'] policy_period = policy['period'] policy_valid_period = policy['valid_period'] policy_exist = True break
                if not policy_exist: raise Exception('購物車中的課程無此價格') policy_course = { 'course_id': course_id, 'course_name': product['name'], 'course_img': product['course_img'], 'policy_id': product['choice_policy_id'], 'policy_price': policy_price, 'policy_period': policy_period, 'policy_valid_period': policy_valid_period, 'coupon_record_list': [ {'id': 0, 'text': '請選擇優惠券'}, ], } policy_course_dict[course_id] = policy_course # 獲取當前全部優惠券
            user_coupon_list = models.CouponRecord.objects.filter(account=request.user, status=0) # ###### 全局優惠券 #######
            global_coupon_record_dict = {} # 課程優惠券添加到課程中；全局優惠券添加到全局
            current_date = datetime.datetime.now().date() for record in user_coupon_list: # 檢查優惠券是否已通過期
                begin_date = record.coupon.valid_begin_date end_date = record.coupon.valid_end_date if begin_date: if current_date < begin_date: continue
                if end_date: if current_date > end_date: continue
                # 全局優惠券
                if not record.coupon.content_type: if record.coupon.coupon_type == 0: temp = {'type': 0, 'text': "通用優惠券", 'id': record.id, 'begin_date': begin_date, 'end_date': end_date, 'money_equivalent_value': record.coupon.money_equivalent_value} elif record.coupon.coupon_type == 1: temp = {'type': 1, 'text': "滿減券", 'id': record.id, 'begin_date': begin_date, 'end_date': end_date, 'minimum_consume': record.coupon.minimum_consume, 'money_equivalent_value': record.coupon.money_equivalent_value} elif record.coupon.coupon_type == 2: temp = {'type': 2, 'text': "折扣券", 'id': record.id, 'begin_date': begin_date, 'end_date': end_date, 'off_percent': record.coupon.off_percent} else: continue global_coupon_record_dict[record.id] = temp # 課程優惠券
                else: cid = record.coupon.object_id if record.coupon.content_type.model == 'course' and cid in policy_course_dict: # 課程價格：滿減，打折，通用
                        if record.coupon.coupon_type == 0: temp = {'type': 0, 'text': "通用優惠券", 'id': record.id, 'begin_date': begin_date, 'end_date': end_date, 'money_equivalent_value': record.coupon.money_equivalent_value} elif record.coupon.coupon_type == 1 and policy_course_dict[cid][ 'policy_price'] >= record.coupon.minimum_consume: temp = {'type': 1, 'text': "滿減券", 'id': record.id, 'begin_date': begin_date, 'end_date': end_date, 'minimum_consume': record.coupon.minimum_consume, 'money_equivalent_value': record.coupon.money_equivalent_value} elif record.coupon.coupon_type == 2: temp = {'type': 2, 'text': "折扣券", 'id': record.id, 'begin_date': begin_date, 'end_date': end_date, 'off_percent': record.coupon.off_percent} else: continue policy_course_dict[cid]['coupon_record_list'].append(temp) user_pay = { 'policy_course_dict': policy_course_dict, 'global_coupon_record_dict': global_coupon_record_dict } redis_pool.conn.hset(settings.REDIS_PAYMENT_KEY, request.user.id, json.dumps(user_pay)) except Exception as e: response['code'] = 1002 response['msg'] = str(e) return Response(response)

05 路飛學城之當即支付

5.1 當即支付 

#!/usr/bin/env python # -*- coding:utf-8 -*-
import json import time import random import datetime from django.conf import settings from django.db import transaction from django.db.models import F from rest_framework.views import APIView from rest_framework.response import Response from api.utils.auth.token_auth import LuffyTokenAuthentication from api.utils.auth.token_permission import LuffyPermission from api.utils import redis_pool from api.utils.alipay import AliPay from repository import models def generate_order_num(): """ 生成訂單編號, 且必須惟一 :return: """
    while True: order_num = time.strftime('%Y%m%d%H%M%S', time.localtime()) + str(random.randint(111, 999)) if not models.Order.objects.filter(order_number=order_num).exists(): break
    return order_num def generate_transaction_num(): """ 生成流水編號, 且必須惟一 :return: """
    while True: transaction_number = time.strftime('%Y%m%d%H%M%S', time.localtime()) + str(random.randint(111, 999)) if not models.TransactionRecord.objects.filter(transaction_number=transaction_number).exists(): break
    return transaction_number class PayOrderView(APIView): authentication_classes = [LuffyTokenAuthentication, ] permission_classes = [LuffyPermission, ] def post(self, request, *args, **kwargs): """ 去支付，生成訂單。 獲取前端提交的購買信息 { course_price_list:[ {'policy_id':1, '':'course_id':1, 'coupon_record_id':1}, {'policy_id':2, '':'course_id':2, 'coupon_record_id':2}, ], coupon_record_id:1, alipay: 99, balance: 1 } 1. 用戶提交 - balance - alipay 2. 獲取去結算列表 課程 3. 循環全部課程 - 獲取原價 - 抵扣的錢 :param request: :param args: :param kwargs: :return: """ response = {'code': 1000} try: # 用戶請求驗證
            policy_course_list = request.data.get('course_price_list') coupon_record_id = request.data.get('coupon_record_id') alipay = request.data.get('alipay')  # >= 0
            balance = request.data.get('balance')  # >= 0

            if balance > request.user.balance: raise Exception('帳戶中貝里餘額不足') # 檢查用戶提交的信息在 redis結算列表 中是否存在，若是不存在，則須要用戶從購物車中再次去結算
            payment_dict_bytes = redis_pool.conn.hget(settings.REDIS_PAYMENT_KEY, request.user.id) payment_dict = json.loads(payment_dict_bytes.decode('utf-8')) policy_course_dict = payment_dict['policy_course_dict'] global_coupon_record_dict = payment_dict['global_coupon_record_dict'] global_coupon_record = {} # 全局優惠券
            if coupon_record_id: if  coupon_record_id not in global_coupon_record_dict: raise Exception('全局優惠券在緩存中不存在') global_coupon_record = global_coupon_record_dict[coupon_record_id] # 當前時間
            current_date = datetime.datetime.now().date() current_datetime = datetime.datetime.now() # 原價
            total_price = 0 # 總抵扣的錢
            discount = 0 # 使用優惠券ID列表
            if coupon_record_id: use_coupon_record_id_list = [coupon_record_id, ] else: use_coupon_record_id_list=[] # 課程和優惠券
            buy_course_record = [] for cp in policy_course_list: _policy_id = cp['policy_id'] _course_id = cp['course_id'] _coupon_record_id = cp['coupon_record_id'] temp = { 'course_id': _course_id, 'course_name': "course", 'valid_period': 0,  # 有效期：30
                    'period': 0,  # 有效期：一個月
                    'original_price': 0, 'price': 0, } if str(_course_id) not in policy_course_dict: raise Exception('課程在緩存中不存在') redis_course = policy_course_dict[str(_course_id)] if str(_policy_id) != str(redis_course['policy_id']): raise Exception('價格策略在緩存中不存在') # 課程是否已經下線或價格策略被修改
                policy_object = models.PricePolicy.objects.get(id=_policy_id)  # 價格策略對象
                course_object = policy_object.content_object  # 課程對象

                if course_object.id != _course_id: raise Exception('課程和價格策略對應失敗') if course_object.status != 0: raise Exception('課程已下線，沒法購買') # 選擇的優惠券是否在緩存中
                redis_coupon_list = redis_course['coupon_record_list'] redis_coupon_record = None for item in redis_coupon_list: if item['id'] == _coupon_record_id: redis_coupon_record = item break
                if not redis_coupon_record: raise Exception('單課程優惠券在緩存中不存在') # 計算購買原總價
                total_price += policy_object.price # 未使用單課程優惠券
                if redis_coupon_record['id'] == 0: temp['price'] = policy_object.price buy_course_record.append(temp) continue temp['original_price'] = policy_object.price temp['valid_period'] = redis_coupon_record['policy_valid_period'] temp['period'] = redis_coupon_record['policy_period'] # 緩存中的優惠券是否已通過期
                begin_date = redis_coupon_record.get('begin_date') end_date = redis_coupon_record.get('end_date') if begin_date: if current_date < begin_date: raise Exception('優惠券使用還未到時間') if end_date: if current_date > end_date: raise Exception('優惠券已過時') # 使用的是單課程優惠券抵扣了多少錢；使用的 我的優惠券ID
                if redis_coupon_record['type'] == 0: # 通用優惠券
                    money = redis_coupon_record['money_equivalent_value'] discount += money elif redis_coupon_record['type'] == 1: # 滿減券
                    money = redis_coupon_record['money_equivalent_value'] minimum_consume = redis_coupon_record['minimum_consume'] if policy_object.price >= minimum_consume: discount += money elif redis_coupon_record['type'] == 2: # 打折券
                    money = policy_object.price * redis_coupon_record['off_percent'] discount += money temp['price'] = policy_object.price - money buy_course_record.append(temp) use_coupon_record_id_list.append(redis_coupon_record['id']) # 全局優惠券
            print(global_coupon_record) begin_date = global_coupon_record.get('begin_date') end_date = global_coupon_record.get('end_date') if begin_date: if current_date < begin_date: raise Exception('優惠券使用還未到時間') if end_date: if current_date > end_date: raise Exception('優惠券已過時') # 使用全局優惠券抵扣了多少錢
            if global_coupon_record.get('type') == 0: # 通用優惠券
                money = global_coupon_record['money_equivalent_value'] discount += money elif global_coupon_record.get('type') == 1: # 滿減券
                money = global_coupon_record['money_equivalent_value'] minimum_consume = global_coupon_record['minimum_consume'] if (total_price - discount) >= minimum_consume: discount += money elif global_coupon_record.get('type') == 2: # 打折券
                money = (total_price - discount) * global_coupon_record['off_percent'] discount += money # 貝里抵扣的錢
            if balance: discount += balance if (alipay + discount) != total_price: raise Exception('總價、優惠券抵扣、貝里抵扣和實際支付的金額不符') # 建立訂單 + 支付寶支付
            # 建立訂單詳細
            # 貝里抵扣 + 貝里記錄
            # 優惠券狀態更新
            actual_amount = 0 if alipay: payment_type = 1  # 支付寶
                actual_amount = alipay elif balance: payment_type = 3  # 貝里
            else: payment_type = 2  # 優惠碼
 with transaction.atomic(): order_num = generate_order_num() if payment_type == 1: order_object = models.Order.objects.create( payment_type=payment_type, order_number=order_num, account=request.user, actual_amount=actual_amount, status=1,  # 待支付
 ) else: order_object = models.Order.objects.create( payment_type=payment_type, order_number=order_num, account=request.user, actual_amount=actual_amount, status=0,  # 支付成功，優惠券和貝里已夠支付
                        pay_time=current_datetime ) for item in buy_course_record: detail = models.OrderDetail.objects.create( order=order_object, content_object=models.Course.objects.get(id=item['course_id']), original_price=item['original_price'], price=item['price'], valid_period_display=item['period'], valid_period=item['valid_period'] ) models.Account.objects.filter(id=request.user.id).update(balance=F('balance') - balance) models.TransactionRecord.objects.create( account=request.user, amount=request.user.balance, balance=request.user.balance - balance, transaction_type=1, content_object=order_object, transaction_number=generate_transaction_num() ) effect_row = models.CouponRecord.objects.filter(id__in=use_coupon_record_id_list).update( order=order_object, used_time=current_datetime) if effect_row != len(use_coupon_record_id_list): raise Exception('優惠券使用失敗') response['payment_type'] = payment_type # 生成支付寶URL地址
                if payment_type == 1: pay = AliPay(debug=True) query_params = pay.direct_pay( subject="路飛學城",  # 商品簡單描述
                        out_trade_no=order_num,  # 商戶訂單號
                        total_amount=actual_amount,  # 交易金額(單位: 元 保留倆位小數)
 ) pay_url = "https://openapi.alipaydev.com/gateway.do?{}".format(query_params) response['pay_url'] = pay_url except IndentationError as e: response['code'] = 1001 response['msg'] = str(e) return Response(response)

06 路飛學城以後續計劃

6.1 後續計劃

07 Flask框架簡介和快速使用

7.1 Flask Web框架；

7.1.1 Flask的上下文管理；

7.1.2 談談你對Python相關的Web框架的理解;

  • Django：大而全，重武器，內部提供：ORM、Admin、中間件、Form、ModelForm、Session、緩存、信號、CSRF；
  • Flask：短小精悍，可拓展強，http://flask.pocoo.org/ 
  • Tornado,短小精悍，可拓展性較之Flask弱一些，但優勢是：異步非阻塞；
  • Web.py：比較老的Web框架；
  • bottle.py：微小，1000行左右；
  • Django的請求過來先走wsgiref，而後middleware；
  • Flask中的WSGI是Werkzurg；

7.2 如何證實Flask內部是Werkzeug

　　7.2.1 Flask的路由是裝飾器；

# -*- coding:utf-8 -*- # Project: FlaskFull  # Software: PyCharm # Time : 2018-09-17 10:12 # File : s2.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


from flask import Flask app = Flask(__name__) @app.route('/index/') def index(): return 'Hello World!'


if __name__ == '__main__': app.run()#run_simple(host,port,app)

08 FLask框架之用戶管理系統示例（一）

8.1 Flask框架之用戶登陸程序；

8.2 使用Pycharm安裝Flask；

from flask import Flask, render_template, request, redirect, session app = Flask(__name__, template_folder='templates') # 基於app這個對象設置secret_key的值，任意設置！
app.secret_key = 'nishifdalkj4389!@#$28908' @app.route('/login/', methods=['GET', 'POST']) def hello_world(): # return 'Hello World!'
    if request.method == "GET": return render_template('login.html') user = request.form.get('usr') pwd = request.form.get('pwd') if user == 'cuixiaozhao' and pwd == '123456': # 將用戶信息放入session;
        session['user_info'] = user """ RuntimeError: The session is unavailable because no secret key was set. Set the secret_key on the application to something unique and secret. 127.0.0.1 - - [17/Sep/2018 21:02:04] "POST /login/ HTTP/1.1" 500 - """
        return redirect('/index/') else: # 兩種傳值方法都可，比Django靈活一些；
        # return render_template('login.html', msg='用戶名或者密碼錯誤!')
        return render_template('login.html', **{'msg': '用戶名或者密碼錯誤！'}) @app.route('/index/') def index(): user_info = session.get('user_info') if not user_info: return redirect('/login/') else: return '歡迎登錄！' @app.route('/logout/') def logout(): del session['user_info'] return redirect('/login/') if __name__ == '__main__': app.run()

09 Flask框架之用戶管理系統示例（二）

9.1 Flask框架用戶登陸示例二； 

from flask import Flask, render_template, request, redirect, session app = Flask(__name__, template_folder='templates') # 基於app這個對象設置secret_key的值，任意設置！
app.secret_key = 'nishifdalkj4389!@#$28908' app.debug = True USER_DICT = { '1': {'name': '志軍', 'age': 18}, '2': {'name': '大偉', 'age': 48}, '3': {'name': '美凱', 'age': 38} } @app.route('/login/', methods=['GET', 'POST']) def hello_world(): # return 'Hello World!'
    if request.method == "GET": return render_template('login.html') user = request.form.get('usr') pwd = request.form.get('pwd') if user == 'cuixiaozhao' and pwd == '123456': # 將用戶信息放入session;
        session['user_info'] = user """ RuntimeError: The session is unavailable because no secret key was set. Set the secret_key on the application to something unique and secret. 127.0.0.1 - - [17/Sep/2018 21:02:04] "POST /login/ HTTP/1.1" 500 - """
        return redirect('/index/') else: # 兩種傳值方法都可，比Django靈活一些；
        # return render_template('login.html', msg='用戶名或者密碼錯誤!')
        return render_template('login.html', **{'msg': '用戶名或者密碼錯誤！'}) @app.route('/index/') def index(): user_info = session.get('user_info') if not user_info: return redirect('/login/') return render_template('index.html', user_dict=USER_DICT) @app.route('/detail/') def detail(): user_info = session.get('user_info') if not user_info: return redirect('/login/') uid = request.args.get('uid') info = USER_DICT.get(uid) return render_template('detail.html', info=info) @app.route('/logout/') def logout(): del session['user_info'] return redirect('/login/') if __name__ == '__main__': app.run()

10 homework

10.1 Flask裝飾器、位置、url起個別名；

10.2 相似於Django中間件的東西，before_request裝飾器；

10.3 上下文管理預習；

  • threading.local;

  • functools.wrappers;

  • functools.partial;

  • 面向對象中——__setarrt__;__getattr__;__delatrr__;

第2章 章節二

01 內容概要

1.1 配置文件；

1.2 路由

1.3 視圖函數

1.4 請求和響應

1.5 templates模板

1.6 session（默認存儲的簽名的cookies中）

1.7 flash閃現

1.8 藍圖blueprint

1.9 常見的裝飾器before_request

1.10 Flask中間件

02 內容回顧

2.1 裝飾器；

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 10:45 # File : 1.裝飾器.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

import functools def wapper(func): @functools.wraps(func) def inner(*args,**kwargs): return func(*args,**kwargs) return inner ''' 一、執行wapper函數，並將被裝飾的函數當作參數。wapper(index) 二、將第一步的返回值，從新賦值給index = wapper（old index） '''

#一、爲何要使用裝飾器？在不改變原來函數的基礎之上，對函數執行先後進行自定義操做；
@wapper def index(a1): return a1 +1000 v = index(2) print(v) #獲取函數名
print("打印函數名:",index.__name__) @wapper def order(a1): return a1+1000

print(index.__name__) print(order.__name__)

2.2 帶參數的裝飾器；

2.3 什麼是面向對象，爲何要使用面向對象？

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 11:06 # File : 3.面向對象.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


""
""" 談談你對面向對象的認識？ 封裝： 將同一類方法分爲一類，方法封裝到類中； 將方法中的共同的參數封裝到對象中，把共同的值封裝到對象中； """


# 用戶類實現;
class File: def __init__(self, a1, a2, a3, a4): self.a1 = a1 self.a2 = a2 self.a3 = a3 self.a4 = a4 def file_add(self): pass

    def file_del(self): pass

    def file_update(self): pass

    def file_fetch(self): pass


# 給了一些值，將數據加工，應用場景:Django自定義分頁;
class Foo(): def __init__(self, a1, a2, a3, a4, a5, a6, a7): self.a1 = a1 self.a2 = a2 self.a3 = a3 self.a4 = a4 self.a5 = a5 self.a6 = a6 self.a7 = a7 def sum(self): return self.a1 + self.a2 def reduce(self): return self.a5 - self.a7 obj = File(1, 2, 3, 4) print(obj)  # <__main__.File object at 0x10bbf25c0>


class A(object): def __init__(self): self.age1 = 123 self.a = A() class B(object): def __init__(self): self.age2 = 123 self.b = B() class C(object): def __init__(self): self.age3 = 123 self.c = C() class D(object): def __init__(self): self.age4 = 123 self.d = D()

03 Flask框架之配置

3.1 app.py；

from flask import Flask, render_template, redirect app = Flask(__name__) # Flask的配置文件這麼玩耍;
app.config.from_object("settings.DevelopmentConfig")#settings後面是一個類名； @app.route('/index/', methods=['GET', 'POST']) def index(): return 'Hello World!'


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

3.2 指定settings.py文件； 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 11:25 # File : settings.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


class BaseConfig(object): DEBUG = False TESTING = False DATABASE_URI = 'sqllite://:memory:'


class ProductionConfig(BaseConfig): DATABASE_URI = 'mysql://user@production/foo'


class DevelopmentConfig(BaseConfig): DEBUG = True DATABASE_URI = 'mysql://user@development/foo'


class TestingConfig(BaseConfig): DEBUG = True DATABASE_URI = 'mysql://user@test/foo'

3.3 Flask配置文件詳解； 

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(...) 方式二： app.config.from_pyfile("python文件名稱") 如： settings.py DEBUG = True app.config.from_pyfile("settings.py") app.config.from_envvar("環境變量名稱") 環境變量的值爲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目錄；

04 Flask框架之建立路由的兩種方式

4.1 基於@app.route('/index/', methods=['GET', 'POST'])裝飾器方式實現；

4.2 經過研究源代碼，基於app.add_url_rule('/order/', view_func=order)實現;

from flask import Flask, render_template, redirect app = Flask(__name__) # Flask的配置文件這麼玩耍;
app.config.from_object("settings.DevelopmentConfig") # 添加的第一種方式,推薦使用裝飾器的方式;
@app.route('/index/', methods=['GET', 'POST']) def index(): return '# 添加的第一種方式,推薦使用裝飾器的方式;'


# 添加路由的另一種方式;


def order(): return '# 添加路由的第二種方式;' app.add_url_rule('/order/', view_func=order) if __name__ == '__main__': app.run() if __name__ == '__main__': app.run()

05 Flask框架之反向生成URL

5.1 Flask框架之反向生成URL，（url_for,endpoint） 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 12:06 # File : 4.反向生成URL.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


from flask import Flask, render_template, redirect, url_for app = Flask(__name__) # endpoint&url_for不起別名默認就是函數名;
@app.route('/index/', methods=['GET', 'POST']) def index(): v1 = url_for('n1') # v2 = url_for('n1')
    # v2 = url_for('n2')
    v2 = url_for('login') v3 = url_for('logout') print(v1, v2, v3) return 'Index' @app.route('/login/', methods=['GET', 'POST'], endpoint='n2') def login(): return 'Login' @app.route('/logout/', methods=['GET', 'POST'], endpoint='n3') def logout(): return 'Logout'

06 Flask框架之自定義路由轉換器

6.1 Flask之自定義路由轉換器； 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 19:14 # File : 6.Flask框架之app.route參數.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask app = Flask(__name__) # 從舊功能重定向至新功能頁面;Js能夠作重定向；meta頭、js location href
@app.route('/index/', methods=['GET', 'POST'], redirect_to='/new/') def index(): return '舊的功能' @app.route('/new/', methods=['GET', 'POST']) def new(): return '新功能'


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

6.2 常見的路由系統；

  • @app.route('/user/<username>')
  • @app.route('/post/<int:post_id>')
  • @app.route('/post/<float:post_id>')
  • @app.route('/post/<path:path>')
  • @app.route('/login', methods=['GET', 'POST']) 
DEFAULT_CONVERTERS = { 'default': UnicodeConverter, 'string': UnicodeConverter, 'any': AnyConverter, 'path': PathConverter, 'int': IntegerConverter, 'float': FloatConverter, 'uuid': UUIDConverter, }

 

def auth(func): def inner(*args, **kwargs): print('before') result = func(*args, **kwargs) print('after') return result return inner @app.route('/index.html',methods=['GET','POST'],endpoint='index') @auth def index(): return 'Index' 或 def index(): return "Index" self.add_url_rule(rule='/index.html', endpoint="index", view_func=index, methods=["GET","POST"]) or app.add_url_rule(rule='/index.html', endpoint="index", view_func=index, methods=["GET","POST"]) app.view_functions['index'] = index 或 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.route和app.add_url_rule參數： rule, URL規則 view_func, 視圖函數名稱 defaults=None,              默認值,當URL中無參數，函數須要參數時，使用defaults={'k':'v'}爲函數提供參數 endpoint=None,              名稱，用於反向生成URL，即： url_for('名稱') methods=None,               容許的請求方式，如：["GET","POST"] strict_slashes=None,        對URL最後的 / 符號是否嚴格要求， 如： @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>') 或 def func(adapter, nid): return "/home/888" @app.route('/index/<int:nid>', redirect_to=func) subdomain=None, 子域名訪問 from flask import Flask, views, url_for app = Flask(import_name=__name__) app.config['SERVER_NAME'] = 'wupeiqi.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" @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() a.註冊路由原理

07 Flask框架之app.route參數

7.1 app.route的常見參數; 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 19:14 # File : 4.Flask框架之app.route參數.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask app = Flask(__name__) # 從舊功能重定向至新功能頁面;Js能夠作重定向；meta頭、js location href
@app.route('/index/', methods=['GET', 'POST'], redirect_to='/new/') def index(): return '舊的功能' @app.route('/new/', methods=['GET', 'POST']) def new(): return '新功能'


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

8 Flask框架之獲取子域名的路由

8.1 域名解析之A記錄；

8.2 本地域名映射之hosts文件；

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 19:14 # File : 7.Flask框架之獲取子域名.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask app = Flask(__name__) app.config['SERVER_NAME'] = 'www.cuixiaozhao.com:5000'


# 從舊功能重定向至新功能頁面;Js能夠作重定向；meta頭、js location href
@app.route('/dynamic/', methods=['GET', 'POST'], subdomain='<username>') def sub_domain(username): print(username) return '舊的功能1'


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

09 上述內容總結

9.1 url;

9.2 methods;

9.3 endpoint;

9.4 @app.route('/index/<int:nid1>/<int:nid2>');

9.5 url_for;

10 Flask框架之視圖中添加裝飾器

10.1 Flask框架中添加裝飾器的注意事項；

  • @裝飾器的順序

  • 引入functools

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 19:14 # File : 8.添加裝飾器.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask import functools app = Flask(__name__) def wapper(func): @functools.wraps(func) def inner(*args, **kwargs): print('before') return func(*args, **kwargs) return inner @app.route('/xxxx/', methods=['GET', 'POST']) @wapper def index(): return 'Index' @app.route('/xxxx/', methods=['GET', 'POST']) @wapper def order(): return 'Order'


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

11 Flask框架之CBV和FBV

11.1 CBV與FBV；

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 21:57 # File : 9.Flask框架之CBV和FBV.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask, redirect, render_template, views app = Flask(__name__) import functools def wapper(func): @functools.wraps(func) def inner(*args, **kwargs): print('before') return func(*args, **kwargs) return inner @app.route('/xxx/', methods=['GET', 'POST']) @wapper def index(): return 'Index'


class IndexView(views.View): methods = ['GET'] decorators = [wapper, ] def dispatch_request(self): print('Index') return 'Index' app.add_url_rule('/index/', view_func=IndexView.as_view(name='index'))  # name == endpoint


# CBV方式;
class IndexView(views.MethodView): methods = ['GET'] decorators = [wapper] 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

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

12 Flask框架之請求和響應相關

12.1 Flask請求相關之request.xxx；

12.2 Flask響應相關之return和response;

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 22:13 # File : 12.請求和響應.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


from flask import Flask, render_template, request, redirect, jsonify, make_response app = Flask(__name__) app.config.from_object("settings.DevelopmentConfig") @app.route('/index/', methods=['GET', 'POST']) def index(): # 請求相關;
 request.args # 響應相關;
    return ''
    return render_template() return redirect('/index/') # 返回json數據;
    return json.dumps({})  # return jsonify({})


if __name__ == '__main__': app.run() """ # 請求相關信息; # request.method # request.args # request.form # 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') # 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 """

13 Flask框架之模板引擎

13.1 Flask的templates模板;

13.2 通Django十分相似，具有block繼承、extends；

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-18 22:23 # File : 13.模板.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask, render_template, redirect, jsonify, make_response, Markup app = Flask(__name__) # 全局模板——每一個模板都可調用的函數;
@app.template_global() def cxz(a1, a2): return a1 + a2 def input(value): return Markup("<input value:'%s'/>" % value) def gen_input(value): return Markup("<input value:'%s'/>" % value) @app.route('/computed/', methods=['GET', 'POST']) def computed(): context = { 'k1': 123, 'k2': [11, 22, 33], 'k3': {'name': 'cuixiaozhao', 'age': 84}, 'k4': lambda x: x + 1,  # 用戶寫簡單的函數;
        'k5': gen_input } return render_template('index.html', **context) if __name__ == '__main__': app.run()

layout.html;

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    <meta name="viewport" content="width=device-width, initial-scale=1">
</head>
<body>
<div>頭部</div>
<div> {% block content %} {% endblock %} </div>
<div>底部</div>
</body>
</html>

indexx.html;

{% extends 'layout.html' %} {% block content %} <h1>{{ k1 }}</h1>
    <h1>{{ k2.0 }} {{ k2[0] }}</h1>
    <h1>{{ k3.name }} {{ k3['name'] }}{{ k3.get('name',19930911) }}</h1>
    <h1>{{ k4 }}</h1>
    <h1>{{ k5(99) }}</h1> {% endblock %}

14 Flask框架之session使用和源碼流程（一）

15 Flask框架之session使用和源碼流程（二）

14.1 & 15.1 之session使用和源碼; 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 16:37 # File : 1.Flask中的session.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
""
""" Session的請求流程; 一、請求剛剛到達; 二、視圖函數; 三、請求結果; """
from flask import Flask, session app = Flask(__name__) app.secret_key = 'fdjljfaljfkla' @app.route('/index/') def index(): session['k1'] = 123
    return 'Index' @app.route('/order/') def order(): print(session['k1']) return 'Order'


if __name__ == '__main__': app.run() """ 一、Flask 二、RequestContext 三、Request 四、SecureCookieSessionInterface 五、SecureCookieSession(dict ) """

16 Flask框架之before_request和after_request(一)

16.1 Flask框架中的內置裝飾器之before_request和after_request; 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 17:22 # File : 1.Flask中內置的特殊裝飾器.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask, render_template, redirect app = Flask(__name__) # before_request和after_request相似於Django中的中間件middleware；
@app.before_request def before_req(): print('before_request，前') @app.before_request def before_req1(): print('before_request1，前') # request以前，會添加一個reverse反轉;
@app.after_request def after_req(response): print('after_request, 後') return response @app.after_request def after_req1(response): print('after_request1, 後') return response @app.route('/x1/', methods=['GET', 'POST']) def x1(): print('視圖函數X1') return 'X1' @app.route('/x2/', methods=['GET', 'POST']) def x2(): print('視圖函數X2') return 'X2'


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

17 Flask框架之after_request(二)

17.1 基於before_request的用戶登陸示例； 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 17:22 # File : 2.基於Flask中內置的特殊裝飾器作登陸驗證.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask, render_template, redirect, request, session app = Flask(__name__) app.secret_key = 'fdsjklfdjaslkjflas'


# before_request和after_request相似於Django中的中間件middleware；
@app.before_request def check_login(): if request.path == '/login/': return None user = session.get('user_info') if not user: return redirect('/login/') @app.route('/login/', methods=['GET', 'POST']) def login(): return 'Login' @app.route('/index/', methods=['GET', 'POST']) def index(): return 'Index'


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

18 Flask框架字號常見裝飾器

18.1 Flask中其餘常見的裝飾器;

  • before_first_request
  • before_request
  • after_request
  • teardown_request
  • after_this_request
  • errorhandler(404)
  • ...
# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 17:44 # File : 3.Flask中其餘常見的裝飾器.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask, Request, render_template app = Flask(__name__, template_folder='templates') app.debug = True @app.before_first_request def before_first_request1(): print('before_first_request1') @app.before_first_request def before_first_request2(): print('before_first_request2') @app.before_request def before_request1(): Request.nnn = 123
    print('before_request1') @app.before_request def before_request2(): print('before_request2') @app.after_request def after_request1(response): print('before_request1', response) return response @app.after_request def after_request2(response): print('before_request2', response) return response @app.errorhandler(404) def page_not_found(error): return 'This page does not exist', 404 @app.template_global() def sb(a1, a2): return a1 + a2 @app.template_filter() def db(a1, a2, a3): return a1 + a2 + a3 @app.route('/') def hello_world(): return render_template('index.html') if __name__ == '__main__': app.run()

19 上述內容總結

19.1小結:

　　19.1.1配置文件

　　19.1.2路由

　　19.1.3視圖之CBV（class-based views ）&FBV（function base view）

　　19.1.4request

　　19.1.5response = make_response(...)

　　19.1.6模板

　　19.1.7session

　　19.1.8常見的裝飾器

20 Flask之閃現

20.1 Flask之消息閃現flask& get_flashed_messages；（內部原理基於session實現）

 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 18:03 # File : 1.Flask之消息閃現flush.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


from flask import Flask, session, flash, get_flashed_messages app = Flask(__name__) app.secret_key = 'fdjslkjflkafdaklfjdlakfj'


# # 生成session; # @app.route('/x1/', methods=['GET', 'POST']) # def login(): # session['mgs'] = 'cuixiaozhao' # return '視圖函數1' # # # # 銷燬session;; # @app.route('/x2/', methods=['GET', 'POST']) # def index(): # msg = session.pop('msg') # print(msg) # return '視圖函數2' # 消息閃現之flask生成session;
@app.route('/x1/', methods=['GET', 'POST']) def login(): flash('cuixiaozhao', category='x1') flash('cuixiaozhao', category='x2') return '視圖函數1'


# 消息閃現之flask銷燬session;;
@app.route('/x2/', methods=['GET', 'POST']) def index(): data = get_flashed_messages(category_filter=['x1', 'x2']) print(data) return '視圖函數2'


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

21 Flask中間件

21.1 Flask中自定義中間件; 

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 18:13 # File : 1.Flask之中間件.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

from flask import Flask app = Flask(__name__) app.secret_key = 'fjaljfdklajfkdasl' @app.route('/x2', methods=['GET', 'POST']) def index(): return 'x2'


class MiddleWare(object): def __init__(self, old_wsgi_app): self.old_wsgi_app = old_wsgi_app def __call__(self, *args, **kwargs): print('before') obj = self.old_wsgi_app(*args, **kwargs) print('after') return obj if __name__ == '__main__': app.wsgi_app = MiddleWare(app.wsgi_app) app.run() """ 一、執行app.__call__方法; 二、在調用app.wsgi_app方法; """

22 Flask框架之藍圖（一）

　　簡單來講，Blueprint 是一個存儲操做方法的容器，這些操做在這個Blueprint 被註冊到一個應用以後就能夠被調用，Flask 能夠經過Blueprint來組織URL以及處理請求。Flask使用Blueprint讓應用實現模塊化，在Flask中，Blueprint具備以下屬性：vue

  • 一個應用能夠具備多個Blueprint；
  • 能夠將一個Blueprint註冊到任何一個未使用的URL下好比 「/」、「/sample」或者子域名；
  • 在一個應用中，一個模塊能夠註冊屢次；
  • Blueprint能夠單獨具備本身的模板、靜態文件或者其它的通用操做方法，它並非必需要實現應用的視圖和函數的；
  • 在一個應用初始化時，就應該要註冊須要使用的Blueprint；

22.1 建立Flask項目ProFlask;

22.2 自定義文件存儲目錄；

__init__.py;

from flask import Flask app = Flask(__name__) # @app.route('/index/') # def index(): # pass


from .views import account from .views import admin from .views import user app.register_blueprint(account.ac) app.register_blueprint(admin.ad) app.register_blueprint(user.us)

accounts.py;

# -*- coding:utf-8 -*- # Project: Pro_Flask  # Software: PyCharm # Time : 2018-09-19 19:08 # File : account.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

from flask import Flask, render_template # 藍圖Blueprint;
from flask import Blueprint ac = Blueprint('ac', __name__) @ac.route('/login/') def login(): #return 'Login'
    return render_template('login.html') @ac.route('/logout/') def logout(): return 'Logout' 

23 Flask框架之藍圖（二）

23.1 藍圖的三大做用（至關於Django中的app）；

  • 目錄結構的劃分；
  • URL的劃分；
  • 給每一類URL添加before_request

24 拓展知識：pipreqs工具

24.1 項目依賴；

24.2 拿到代碼啓動不起來；

24.3 pip3 install pipreqs;（自動查找項目所須要的依賴包及版本號）；

24.4 pipreqs ./ --force 強制生成依賴文件；

24.5 pip3 install -i requirements.txt 安裝依賴文件；

25 拓展知識：函數和方法

25.1 什麼是函數？

25.2 什麼是方法？

# -*- coding:utf-8 -*- # Project: Day123  # Software: PyCharm # Time : 2018-09-19 20:40 # File : 1.函數和方法的區別.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


from types import MethodType, FunctionType class Foo(object): def fetch(self): pass


print(isinstance(Foo.fetch, MethodType))  # False
print(isinstance(Foo.fetch, FunctionType))  # True
 obj = Foo() print(obj.fetch)  # <bound method Foo.fetch of <__main__.Foo object at 0x10bbf2358>>
print(isinstance(obj.fetch, MethodType))  # False
print(isinstance(Foo.fetch, FunctionType))  # True
""" 類、對象、方法、函數能夠➕()； 方法和函數的區別：被誰來調用！ """

第3章 章節三

01 內容概要

1.1 Flask提高逼格的時候；

1.2 Flask的上下文管理；

1.3 threading.local;

1.4 數據庫鏈接池;

02 內容回顧

2.1 經常使用的Linux命令（100+）；

2.2 常見算法搞定；

2.3 數據庫鏈接池；

2.4 面向對象的特殊方法；

  • call;
  • new;
  • and;
  • equal;
  • next;
  • dict;

2.5 functools；

  • functools.partial();
  • 裝飾器的應用場景；-Flask路由以及before_request、登陸認證、Django緩存\CSRF_TOKEN；

2.6 Flask中的藍圖Blueprint;

2.7 Flask中的session;

03 threading.local學習

3.1 threading.local初識；

# -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-19 22:40 # File : 1.ThreadingLocal.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

from threading import local from threading import Thread import time # 特殊的對象;
xiaozhao = local() # xiaozhao = -1


def task(arg): # global xiaozhao
 xiaozhao.value = arg time.sleep(2) print(xiaozhao.value) for i in range(10): t = Thread(target=task, args=(i,)) t.start()

04 自定義Local思路

05 自定義Local對象（基於函數）

5.1 基於get_ident實現；

 

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-20 09:11 # File : 3.自定義Local對象（基於函數）.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

from threading import get_ident, Thread import time storage = {} def set(k, v): ident = get_ident() if ident in storage: storage[ident][k] = v storage[ident] = {k: v} def get(k): ident = get_ident() return storage[ident][k] def task(arg): set('val', arg) print(storage) time.sleep(2) v = get('val') print(v) for i in range(10): t = Thread(target=task, args=(i,)) t.start()

06 自定義Local對象（基於面向對象）

6.1 基於面向對象basic；

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-20 09:20 # File : 4.自定義Local對象(基於面向對象).py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from threading import get_ident from threading import Thread class Local(object): storage = {} def set(self, k, v): ident = get_ident() if ident in Local.storage: Local.storage[ident][k] = v else: Local.storage[ident] = {k: v} def get(self, k): ident = get_ident() return Local.storage[ident][k] obj = Local() def task(arg): obj.set('val', arg) v = obj.get('val') print(v) for i in range(10): t = Thread(target=task, args=(i,)) t.start()

6.2 基於面向對象優化版；

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-20 09:20 # File : 5.自定義Local對象(基於面向對象優化版 ).py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from threading import get_ident from threading import Thread class Local(object): def __setattr__(self, k, v): # self.storage = {}
        object.__setattr__(self, 'storage', {}) ident = get_ident() if ident in self.storage: self.storage[ident][k] = v else: self.storage[ident] = {k: v} def __getattr__(self, item): ident = get_ident() return self.storage[ident][item] obj = Local() obj1 = Local() def task(arg): obj.val = arg obj1.val = arg print(obj.val) for i in range(10): t = Thread(target=task, args=(i,)) t.start()

6.3 基於面向對象升級版；

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-20 09:20 # File : 4.自定義Local對象(基於面向對象).py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from threading import get_ident from threading import Thread class Local(object): storage = {} def __setattr__(self, k, v): ident = get_ident() if ident in Local.storage: Local.storage[ident][k] = v else: Local.storage[ident] = {k: v} def __getattr__(self, item): ident = get_ident() return Local.storage[ident][item] obj = Local() def task(arg): obj.val = arg print(obj.val) for i in range(10): t = Thread(target=task, args=(i,)) t.start()

6.4 基於面向對象greenlet版本； 

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-20 09:20 # File : 7.自定義Local對象(基於greenlet).py # Author : 天晴天朗 # Email : tqtl@tqtl.org

try: from greenlet import getcurrent as get_ident except Exception as e: from threading import get_ident from threading import Thread class Local(object): storage = {} def __setattr__(self, k, v): ident = get_ident() if ident in Local.storage: Local.storage[ident][k] = v else: Local.storage[ident] = {k: v} def __getattr__(self, item): ident = get_ident() return Local.storage[ident][item] obj = Local() def task(arg): obj.val = arg print(obj.val) for i in range(10): t = Thread(target=task, args=(i,)) t.start()

07 Flask上下文管理之本質分析 

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Threading  # Software: PyCharm # Time : 2018-09-20 09:45 # File : 1.Flask源碼分析上下文管理.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask app = Flask(__name__) @app.route('/') def hello_world(): return 'cuixiaozhao!'


if __name__ == '__main__': app.__call__ app.run() ""
""" 一、第一個階段：將ctx(request，session)放到Local對象; 二、第二階段:視圖函數導入：request、session； 三、請求處理完畢： -獲取session並保存到cookie； -將ctx刪除； """

08 Flask上下文管理之請求處處理階段

8.1 Flask中的session是什麼時候建立什麼時候銷燬的？request_context，localstack、local；

09 Flask上下文管理之視圖調用階段

9.1 調用階段；

10 Flask上下文管理之視圖調動階段和結束階段

10.1 視圖調用階段和結束階段；

11 問題來了

11.1 Flask中一共有幾個Local和Local對象？

都是2個；

12 Flask中的g究竟是什麼呢？生命週期？

12.1 Flask的g對象範圍；

13 內容補充：面向對象的私有字段

13.1 私有字段不建議去調動；

  • obj = Foo()
  • obj._Foo_age(類名前加下劃線)

14 homework

14.1 按照組爲單位，畫圖-類的調用關係圖；

第4章 章節四

01 內容概要

1.1 flask-session；

1.2 單獨模塊-數據庫鏈接池DBUtils；

1.3 原生SQL（基於pymysql）仍是ORM好？！

1.4 wtforms（任何Django框架均可使用）；

1.5 SQLAchemy、flask-sqlachemy；

1.6 flask-script；

1.7 flask-migrate；

02 內容回顧

2.1 談談Django和Flask的認識？

2.2 Django的上下文管理機制？

2.3 ctx = RequestContext(request,session)

2.4 Local對象的做用？

  • 看過Local源碼，和Threading.local類似但又有不一樣之處；
  • 不一樣之處在於，Local中基於greenlet獲取惟一表示，顆粒度更細；

2.5 爲何使用Localstack？對Local對象中的數據進行操做。

2.6 上下文管理分爲兩個-請求上下文和App上下文

2.7 什麼是g？一次請求週期內的全局變量。

2.8 獲取session和g的流程。

2.9 Flask中的技術點。

  • 反射；
  • 面向對象-封裝、繼承和多態；
  • __dict__；
  • 線程相關的東西-threading.local；
  • 本身寫一個類+列表，實現一個棧（基於Localstack實現棧）；

03 路飛學城補充：視頻播放受權

4.1 CC視頻播放受權；

04 flask-session組件使用和原理（一）

 4.1 flask-session；

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 16:14 # File : 1.flask_session.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask, session from flask_session import RedisSessionInterface app = Flask(__name__) app.secret_key = 'fdahfdafdajfalk'

# 默認session保存操做； # from flask.sessions import SecureCookieSessionInterface

# app.session_interface = SecureCookieSessionInterface()

# 使用Redis保存session;
from flask.ext.session import Session from redis import Redis app.config['SESSION_TYPE'] = 'redis' app.config['SESSION_REDIS'] = Redis(host='192.168.0.94', port='6379') app.session_interface = RedisSessionInterface( redis=Redis(host='127.0.0.1', port=6379), key_prefix='flaskxxxx' ) @app.route('/login/') def login(): session['k1'] = 123
    return 'Login' @app.route('/index/') def index(): v = session['k1'] print(v) return 'Index'


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

05 flask-session組件使用和原理（二）

5.1 原理難，實現的機制難，使用起來比較容易，可是要想學好， 必須懂原理；

5.2 MTV 和MVC設計模式；

  • MTV-Model、Templates、View；
  • MVC-Model、View、Controller；

5.3 自定義Flask項目Flask_Session；

__init__.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 16:46 # File : __init__.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask from .views import account from .views import home def create_app(): app = Flask(__name__) app.config.from_object('settings.DevelopmentConfig') app.register_blueprint(account.account) app.register_blueprint(home.home) return app

manage.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 16:46 # File : manage.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

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

account.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 16:53 # File : account.py # Author : 天晴天朗 # Email : tqtl@tqtl.org

from flask import Blueprint, render_template, request, session, redirect from uuid import uuid4 account = Blueprint('account', __name__) @account.route('/login/', methods=['GET', 'POST']) def login(): if request.method == "GET": return render_template("login.html") user = request.form.get('user') pwd = request.form.get('pwd') if user == "cxz" and pwd == "123": uid = str(uuid4()) session.permanent = True session['user_info'] = {'id': uid, 'name': user} return redirect('/index/') else: return render_template('login.html', msg='用戶名或者密碼錯誤！')

home.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 17:33 # File : home.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


from flask import Blueprint, render_template, request, session, redirect home = Blueprint('home', __name__) @home.route('/index/', methods=['GET', 'POST']) def index(): user_info = session.get('user_info')  # {'k1':1,'k2':2}
    print("原來的值", user_info) session['user_info']['k1'] = 19939 user_info = session.get('user_info') print("修改以後的值", user_info) # session['modified'] = True,在配置文件中使用SESSION_REFRESH_EACH_REQUEST代替;
    return 'Index' @home.route('/test/') def test(): user_info = session.get('user_info') print(user_info) return 'Test'

settings.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 17:28 # File : settings.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from datetime import timedelta class Config(object): DEBUG = True TESTING = False DATABASE_URI = 'sqlite://:memory' SECRET_KEY = 'fjdksjfdasljflksd' PERMANENT_SESSION_LIFETIME = timedelta(minutes=20) SESSION_REFRESH_EACH_REQUEST = True class ProductionConfig(Config): pass


class DevelopmentConfig(Config): pass


class TestingConfig(Config): pass

06 flask-session組件使用和原理（三）

6.1 將數據存儲至Redis；

settings.py;html5

 

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 17:28 # File : settings.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from datetime import timedelta from redis import Redis class Config(object): DEBUG = True TESTING = False DATABASE_URI = 'sqlite://:memory' SECRET_KEY = 'fjdksjfdasljflksd' PERMANENT_SESSION_LIFETIME = timedelta(minutes=20) SESSION_REFRESH_EACH_REQUEST = True SESSION_TYPE = "redis"
    # SESSION_REDIS = Redis(host='127.0.0.1',port='6379')


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

 

__init__.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 16:46 # File : __init__.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
from flask import Flask from .views import account from .views import home # from flask.ext.session import Session
from flask_session import Session def create_app(): app = Flask(__name__) app.config.from_object('settings.DevelopmentConfig') app.register_blueprint(account.account) app.register_blueprint(home.home) # 將Session替換成Redis;
 Session(app) return app

 

6.2 Mac版本的Redis可視化工具redis desktop manager的使用；

07 flask-session組件使用和原理（四）

7.1 flask-session的做用：將默認保存的簽名cookie中的值保存到Redis、memcached、file、MongoDB、SQLAchemy；

7.2 配置方法；

app.config['SESSION_TYPE'] = 'redis'python

app.config['SESSION_REDIS'] = Redis(host = '127.0.0.1',port=6379)mysql

7.3 替換方式；

from flask_session import Sessionios

Session(app)web

7.4 注意事項：session中存儲的是字典，修改字典內部元素時候，會形成數據不更新；

  • modified = True ，默認值爲False
  • SESSION_REFRESH_EACH_REQUEST = True and session.permanent = True (Redis中默認)

08 基於pymysql實現用戶登陸

8.1 基於pymysql；

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 20:13 # File : sql.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
import pymysql class SQLHelper(object): @staticmethod def open(): conn = pymysql.connect(host='mysql123.cuixiaozhao.com', port=3306, user='root', passwd='Tqtl911!@%*)', db='flask_session') cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) return conn, cursor @staticmethod def close(conn, cursor): conn.commit() cursor.close() conn.close() @classmethod def fetch_one(cls, sql, args): # cursor.execute("select id,name from users where name = %s and pwd = %s", ['cxz', '123', ])
        conn, cursor = cls.open() cursor.execute(sql, args) obj = cursor.fetchone() cls.close(conn, cursor) return obj @classmethod def fetch_all(cls, sql, args): conn, cursor = cls.open() cursor.execute(sql, args) obj = cursor.fetchall() cls.close(conn, cursor) return obj

8.2 pymysql的練習和使用； 

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-20 19:53 # File : MySQL數據庫練習.py # Author : 天晴天朗 # Email : tqtl@tqtl.org


import pymysql conn = pymysql.connect(host='x.x.x.x', port=3306, user='root', passwd='Tqtl911!@%*)', db='flask_session') cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) cursor.execute("select id,name from users where name = %s and pwd = %s", ['cxz', '123', ]) obj = cursor.fetchone() conn.commit() cursor.close() conn.close() print(obj)

09 基於數據庫鏈接池實現用戶登陸(出bug)

9.1 DBUtils模塊初識；

　　DBUtils是Python的一個用於實現數據庫鏈接池的模塊。面試

pool.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session  # Software: PyCharm # Time : 2018-09-21 10:26 # File : pool.py # Author : 天晴天朗 # Email : tqtl@tqtl.org
import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection from manage import app POOL = PooledDB( creator=pymysql,  # 使用連接數據庫的模塊；
    maxconnections=6,  # 鏈接池容許的最大鏈接數，0和None表示不限制鏈接數；
    mincached=2,  # 初始化時，連接池中至少建立的空閒的連接，0表示不建立；
    maxcached=5,  # 連接池中最多閒置的連接；
    maxshared=3, # 連接池中最多共享的連接數量，0和None表示所有共享。PS: Useless無用，由於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='mysql.cuixiaozhao.com', port=3306, user='root', password='Tqtl911!@%*)', database='flask_session', charset='utf8' )

9.2 模式一：模式一：爲每一個線程建立一個鏈接，線程即便調用了close方法，也不會關閉，只是把鏈接從新放到鏈接池，供本身線程再次使用。當線程終止時，鏈接自動關閉。

POOL = PersistentDB( creator=pymysql,  # 使用連接數據庫的模塊
    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
    closeable=False, # 若是爲False時， conn.close() 實際上被忽略，供下次使用，再線程關閉時，纔會自動關閉連接。若是爲True時， conn.close()則關閉連接，那麼再次調用pool.connection時就會報錯，由於已經真的關閉了鏈接（pool.steady_connection()能夠獲取一個新的連接）
    threadlocal=None,  # 本線程獨享值得對象，用於保存連接對象，若是連接對象被重置
    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()

9.3 模式二：模式二：建立一批鏈接到鏈接池，供全部線程共享使用。

import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection 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='123', database='pooldb', charset='utf8' ) def func(): # 檢測當前正在運行鏈接數的是否小於最大連接數，若是不小於則：等待或報raise TooManyConnections異常
    # 不然
    # 則優先去初始化時建立的連接中獲取連接 SteadyDBConnection。
    # 而後將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中並返回。
    # 若是最開始建立的連接沒有連接，則去建立一個SteadyDBConnection對象，再封裝到PooledDedicatedDBConnection中並返回。
    # 一旦關閉連接後，鏈接就返回到鏈接池讓後續線程繼續使用。
    conn = POOL.connection() # print(th, '連接被拿走了', conn1._con)
    # print(th, '池子裏目前有', pool._idle_cache, '\r\n')
 cursor = conn.cursor() cursor.execute('select * from tb1') result = cursor.fetchall() conn.close() func()

9.4 補充說明; 若是沒有鏈接池，使用pymysql來鏈接數據庫時，單線程應用徹底沒有問題，但若是涉及到多線程應用那麼就須要加鎖，一旦加鎖那麼鏈接勢必就會排隊等待，當請求比較多時，性能就會下降了。

10 解決bug（一）

11 解決bug（二）

12 數據庫鏈接池總結

12.1 每個線程建立一個鏈接，關閉（默認不關閉），線程終止時，才關閉鏈接；

12.2 建立共享鏈接池；

12.3 應用；只要寫原生sql，使用了pymysql就得使用數據庫鏈接池DBUtils；

13 WTforms介紹以及用戶登陸示例

13.1 WTform安裝；

13.2 WTform的使用；

 

#!/usr/bin/env python # -*- coding:utf-8 -*-
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 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() app.py

 

index.html;

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

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

14 WTforms用戶註冊示例

14.1 基於WTforms作用戶註冊驗證；

app.py;

#!/usr/bin/python3 # -*- coding:utf-8 -*- # Project: Flask_Session # Software: PyCharm # Time : 2018-09-21 15:11 # File : register.py # Author : 天晴天朗 # Email : tqtl@tqtl.org from flask import Flask, render_template, request, redirect, Blueprint 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 ) 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': 1}) 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() # app.py

register.html;

<!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 item in form %} <p>{{item.label}}: {{item}} {{item.errors[0] }}</p> {% endfor %} <input type="submit" value="提交">
</form>
</body>
</html>

15 內容總結

15.1 WTforms總結；

WTForms是一個支持多個web框架的form組件，主要用於對用戶請求數據進行驗證。redis

16 homework

16.1 補充了路飛學城CC受權播放；

16.2 flask-session的用法以及Redis的使用；

16.3 pymysql以及數據庫鏈接池DBUtils；

16.4 WTforms；

16.5 Flask的上下文管理以及源碼

16.6 flask-session；

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程