您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

Python-Flask+Mysql搭建後台接口實例

編輯：Python

SQLAlchemy是一個關系型數據庫框架，它提供了高層的 ORM 和底層的原生數據庫的操作。

flask-sqlalchemy是一個簡化了SQLAlchemy操作的flask擴展。

ORM（Object-Relation Mapping 稱為對象-關系映射）：

主要實現模型對象到關系數據庫數據的映射，比如:把數據庫表中每條記錄映射為一個模型對象

ORM提供了一種持久化模式，可以高效地對數據庫進行訪問。它可以把底層的 RDBMS 封裝成業務實體對象，提供給業務邏輯層使用。程序員往往關注業務邏輯層面，而不是底層數據庫該如何訪問，以及如何編寫 SQL 語句獲取數據等等。采用 ORM，就可以從數據庫的設計層面轉化成面向對象的思維。

環境：window11+pycharm2020.1+Anaconda4.11.0 +python3.7

Flask-sqlalchemy2.5.1

Flask-Migrate：3.1.0

源代碼：https://github.com/sueleeyu/flask_api_ar

一、flask-sqlalchemy安裝

1.安裝pymysql，打開控制台，輸入：

pip install pymysql

2.安裝flask-sqlalchemy，打開控制台，輸入：

pip install flask-sqlalchemy

3.安裝Flask-Migrate，打開控制台，輸入：

pip install flask-migrate

解決Flask-migrate安裝和遇到的問題：Python入門-Flask-migrate安裝和使用：

Python入門-Flask-migrate安裝和使用_suelee_hm的博客-CSDN博客

二、相關概念

flask-sqlalchemy操作數據庫，使用起來比較簡單，易於操作。常用的配置如下：

常用的SQLAlchemy字段類型：

類型名

python中類型

說明

Integer

int

普通整數，一般是32位

SmallInteger

int

取值范圍小的整數，一般是16位

BigInteger

int或long

不限制精度的整數

Float

float

浮點數

Numeric

decimal.Decimal

普通整數，一般是32位

String

str

變長字符串

Text

str

變長字符串，對較長或不限長度的字符串做了優化

Unicode

unicode

變長Unicode字符串

UnicodeText

unicode

變長Unicode字符串，對較長或不限長度的字符串做了優化

Boolean

bool

布爾值

Date

datetime.date

時間

Time

datetime.datetime

日期和時間

LargeBinary

str

二進制文件

常用的SQLAlchemy列選項：

選項名

說明

primary_key

如果為True，代表表的主鍵

unique

如果為True，代表這列不允許出現重復的值

index

如果為True，為這列創建索引，提高查詢效率

nullable

如果為True，允許有空值，如果為False，不允許有空值

default

為這列定義默認值

常用的SQLAlchemy關系選項：

選項名

說明

backref

在關系的另一模型中添加反向引用

primary join

明確指定兩個模型之間使用的聯結條件

uselist

如果為False，不使用列表，而使用標量值

order_by

指定關系中記錄的排序方式

secondary

指定多對多關系中關系表的名字

secondary join

在SQLAlchemy中無法自行決定時，指定多對多關系中的二級聯結條件

三、編寫python

1.創建db對象。新建exts.py文件：

# 存放db變量 from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy()

2.創建配置文件。新建configs.py文件：

# -*- coding: UTF-8 -*- # 配置文件 # 數據庫信息 HOST = 'xxx.cn' PORT = '3306' DATABASE = 'db' USERNAME = 'root' PASSWORD = 'root' DB_URI = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE) # 數據庫連接編碼 DB_CHARSET = "utf8" SQLALCHEMY_DATABASE_URI = DB_URI SQLALCHEMY_TRACK_MODIFICATIONS = False SQLALCHEMY_ECHO = True

3.綁定app。新建app.py，編寫代碼創建app，注冊藍圖：

# app.py # 主app文件，運行文件 from flask import Flask from flask_migrate import Migrate from controller import ar import configs from exts import db def create_app():     app1 = Flask(__name__)     # 注冊藍圖     app1.register_blueprint(ar, url_prefix='/ar')     # 加載配置文件     app1.config.from_object(configs)     # db綁定app     db.init_app(app1)     # 要讓Flask-Migrate能夠管理app中的數據庫，需要使用Migrate(app,db)來綁定app和數據庫。假如現在有以下app文件     # 綁定app和數據庫     migrate = Migrate(app=app1, db=db)     return app1 app = create_app()

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

4.創建表。新建models.py,編寫每個數據庫表類:

# models.py # 模型文件，用來存放所有的模型 from exts import db """ 以下表關系： 一個用戶對應多個（一對多）
 """ """ 一對一關系中，需要設置relationship中的uselist=Flase，其他數據庫操作一樣。 一對多關系中，外鍵設置在多的一方中，關系（relationship）可設置在任意一方。 多對多關系中，需建立關系表，設置 secondary=關系表
 """ if __name__ == '__main__':     import app     app1 = app.create_app()     app1.run() # 用戶表 class ARUser(db.Model):  # User 模型名     __tablename__ = 'ar_user'  # 表名     id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)  # id，主鍵自增     user_name = db.Column(db.String(20), index=True)  # 用戶名稱     create_time = db.Column(db.TIMESTAMP)     update_time = db.Column(db.TIMESTAMP)     # relationship     # 1.第一個參數是模型的名字，必須要和模型的名字一致     # 2.backref（bsck reference）：代表反向引用，代表對方訪問我的時候的字段名稱     geospatials = db.relationship('GeoHistory', backref='ARUser', lazy='select')  # 添加關系     def __init__(self, name):         self.user_name = name # Geospatial錨點表 class GeoHistory(db.Model):  # GeoHistory 模型名     __tablename__ = 'ar_geo_history'  # 表名     id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)  # id，主鍵自增     bid = db.Column(db.String(20), index=True)  # geo id     # 外鍵     # 1.外鍵的數據類型一定要看所引用的字段類型，要一樣     # 2. db.Foreignkey("表名. 字段名")fl     # 3.外鍵是屬於數據庫層面的，不推薦直接在ORM直接使用     uid = db.Column(db.BigInteger, db.ForeignKey('ar_user.id'))  # 用戶 id，設置外鍵     name = db.Column(db.String(20), index=True)  # 錨點名稱     latitude = db.Column(db.Float)     longitude = db.Column(db.Float)     altitude = db.Column(db.Float)     heading = db.Column(db.Float)  # 手機攝像頭朝向     state = db.Column(db.SmallInteger)  # 錨點狀態 0隱藏，1顯示     create_time = db.Column(db.TIMESTAMP)     update_time = db.Column(db.TIMESTAMP)     def __init__(self, bid, name, latitude, longitude, altitude, heading):         self.bid = bid         self.name = name         self.latitude = latitude         self.longitude = longitude         self.altitude = altitude         self.heading = heading

db.relationship和backref的用處：

db.relationship是和db.ForeignKey配合使用的，用來描述一對多關系。
在一對多中，db.ForeignKey寫在“多”方，關聯“一”方的某一個屬性。
db.relationship寫在“一”方，“一”方通過該屬性可以取出一個列表，列表元素為所有對應的“多”方的對象。
db.relationship中的backref是“多”方使用的。“多”方通過該屬性（即backref傳入的字符串）可以訪問到其對應的“一”方對象。

5.映射表。創建好表後需要映射到數據庫中，這裡需要用到flask-migrate庫。

app.py中的Migrate綁定代碼:

# db綁定app db.init_app(app) # 要讓Flask-Migrate能夠管理app中的數據庫，需要使用Migrate(app,db)來綁定app和數據庫。假如現在有以下app文件 # 綁定app和數據庫 migrate = Migrate(app=app, db=db)

pycharm進入控制台：

執行init命令，初始化一個遷移文件夾：

flask db init

執行migrate命令，把當前的模型添加到遷移文件中：

flask db migrate

執行update命令，把遷移文件中對應的數據庫操作，真正的映射到數據庫中:

flask db upgrade

執行完畢，數據庫中會生成和models.py中對應的表：

四、POST和GET

Request請求有GET和POST兩種，flask獲取參數的方式：

1. request.form.get("key", type=str, default=None) # 獲取表單數據
2. request.args.get("key") # 獲取get請求參數
3. request.values.get("key") # 獲取所有參數

route裝飾器通過methods指定請求方式，默認為GET：

@app.route('/ar/add_anchor', methods=['POST'])

GET：參數在url中，可用以下兩種方式獲取參數：

1. request.args.get("key") # 獲取get請求參數
2. request.values.get("key") # 獲取所有參數

POST：

POST請求的發送通過Content-Type標記不同數據。

application/json ：請求body體的內容json如{"a": "b", "c": "d"}
application/x-www-form-urlencoded 請求body體如 a=b&c=d

不同Content-Type的處理方式：

a. application/json：

請求：

接收：

jdata = request.get_json()  # 或者 request.json.get('content')

b. application/x-www-form-urlencoded：

請求：

接收：

jdata = request.values.get("content")

c. multipart/form-data

請求：

接收：

jdata = request.form.get('content')

五、數據庫操作

1.插入：

geoHistory = GeoHistory("100001", "bp1", 39.4632, 116.3679, 28.3135, 137.1354) db.session.add(geoHistory) db.session.commit()

批量插入：

geoHistory = GeoHistory("100001", "bp1", 39.4632, 116.3679, 28.3135, 137.1354) geoHistory2 = GeoHistory("100001", "bp2", 39.4632, 116.3679, 28.3135, 137.1354) db.session.add(geoHistory)  # 插入一個 db.session.commit() anchors = [geoHistory, geoHistory2] # 批量 db.session.execute(GeoHistory.__table__.insert(), anchors)  # SQLAlchemy Core db.session.commit() # 批量 or db.session.add_all(anchors) db.session.commit()

2.查詢

Filter進行過濾，filter內條件為and。limit限制數據：

latitude_min, latitude_max, longitude_min, longitude_max = utils.get_area(latitude, longitude, 1000) anchors = GeoHistory.query.filter(     GeoHistory.latitude.between(latitude_min, latitude_max),     GeoHistory.longitude.between(longitude_min, longitude_max)).limit(     20).all()

六、編寫接口

1.入口

@app.route('/') def hello_world():        return 'Hello World!'

2.app內注冊藍圖：

# 注冊藍圖 app1.register_blueprint(ar, url_prefix='/ar')

3.新建controller.py,創建藍圖，編寫接口:

from flask import Blueprint, request from R import R from service import ARService ar = Blueprint("ar", __name__) @ar.route('/nearby') def nearby():     latitude = float(request.args.get('latitude'))     longitude = float(request.values.get('longitude'))     result = ARService.query_histories(latitude, longitude)     return R.ok(result) @ar.route('/add_anchors', methods=["GET", "POST"]) def add_anchor():     json_data = ''     if request.method == "GET":         json_data = request.args.get("content")     if request.method == "POST":         if request.content_type.startswith('application/json'):             json_data = request.get_json()             # application/json 獲取的原始參數，接受的是type是'bytes’的對象，如：b{'name':'lucy', 'age':22}             # data = request.get_data()         elif request.content_type.startswith('multipart/form-data'):             json_data = request.form.get('content')         else:             json_data = request.values.get("content")     anchors = json_data["collection"]     ARService.add_anchor(anchors)     return R.ok(data=None)

4.新建service.py,實現數據邏輯：

import utils from models import GeoHistory from exts import db from utils import O2d class ARService(object):     @staticmethod     def query_histories(latitude, longitude):         # 查詢所有         latitude_min, latitude_max, longitude_min, longitude_max = utils.get_area(latitude, longitude, 1000)         anchors = GeoHistory.query.filter(             GeoHistory.latitude.between(latitude_min, latitude_max),             GeoHistory.longitude.between(longitude_min, longitude_max)).limit(             20).all()         return O2d.obj_to_list(anchors)     @staticmethod     def add_anchor(anchors):         db.session.execute(GeoHistory.__table__.insert(), anchors)  # SQLAlchemy Core         db.session.commit() if __name__ == '__main__':     geoHistory = GeoHistory("100001", "bp1", 39.4632, 116.3679, 28.3135, 137.1354)     geoHistory2 = GeoHistory("100001", "bp2", 39.4632, 116.3679, 28.3135, 137.1354)

測試Json數據：

{ "collection":[{ "bid":"100001","name":"bp1","serializedTime":"2022/7/6 16:43:42","latitude":39.90866127654739,"longitude":116.46562960310614,"altitude":29.459830418229104,"heading":137.04320866867753}]}

Content-Type：application/json

utils.py:

class O2d:     @staticmethod     def obj_to_dic(obj):         '''         將傳入的data對象轉成字典
         '''         result = {}         for temp in obj.__dict__:             if temp.startswith('_') or temp == 'metadata':                 continue             result[temp] = getattr(obj, temp)         return result     @staticmethod     def obj_to_list(list_obj):         '''         將傳入的data對象轉成List,list中的元素是字典
         '''         result = []         for obj in list_obj:             result.append(O2d.obj_to_dic(obj))         return result