安阳网站推广,珠海市网站建设怎么样,交通网上服务平台,51自学网官方网站Python Flask 数据库开发 引言环境配置创建 Flask 应用#xff0c;连接数据库定义路由定义模型创建表创建 API 数据库直接操作启动 Flask 应用app.py 示例运行 Flask访问应用 展望 引言
在现代 web 开发中#xff0c;Python 的 Flask 框架因其轻量和灵活性受到广泛欢迎。结合… Python Flask 数据库开发 引言环境配置创建 Flask 应用连接数据库定义路由定义模型创建表创建 API 数据库直接操作启动 Flask 应用app.py 示例运行 Flask访问应用 展望 引言
在现代 web 开发中Python 的 Flask 框架因其轻量和灵活性受到广泛欢迎。结合数据库技术Flask 可以高效地管理和处理数据使开发者能够快速构建功能强大的应用程序。无论是选择 MySQL 还是 PostgreSQL掌握数据库与 Flask 的集成至关重要。本文将探讨如何在 Flask 中设置和使用数据库涵盖从环境配置到基本操作的各个方面便于我们去实现数据驱动的应用。
环境配置
在开始之前需要确保 Python 环境已安装 Flask 模块。可以通过 pip 安装 Flask 和 SQLAlchemyFlask 的 ORM 工具
pip install Flask Flask-SQLAlchemy接下来根据所选数据库的类型如 MySQL 或 PostgreSQL部署相应的数据库服务。 详情可参考MySQL PostgreSQL 数据库部署
创建 Flask 应用连接数据库
from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp Flask(__name__)
# 创建了一个 Flask 应用实例。__name__ 是 Python 的一个特殊变量它指向当前模块的名字。app.config[SQLALCHEMY_DATABASE_URI] mysql://user:passwordlocalhost/dbname
# mysql://user:passwordlocalhost/dbname 的格式是
# mysql数据库类型。
# user数据库用户名。
# password数据库密码。
# localhost数据库服务器地址这里是本地。
# dbname要连接的数据库名称。db SQLAlchemy(app)
# 创建了一个 SQLAlchemy 对象并将 Flask 应用实例传递给它。这样SQLAlchemy 就可以使用 Flask 应用的配置来管理数据库操作# app.route 是 Flask 中的装饰器用于定义路由。它将一个 URL 路径与一个视图函数关联起来。简单来说当用户访问指定的 URL 时Flask 会调用对应的视图函数并返回结果
app.route(/)
def home():return Hello, Flask!if __name__ __main__:app.run(debugTrue)定义路由
app.route 是 Flask 中的装饰器用于定义路由。它将一个 URL 路径与一个视图函数关联起来。简单来说当用户访问指定的 URL 时Flask 会调用对应的视图函数并返回结果。例如
app.route(/)
def home():return Hello, Flask!在这个例子中访问根 URL (/) 时用户会看到 “Hello, Flask!” 的消息。通过使用不同的路径和方法如 GET 或 POST我们可以创建丰富的 Web 应用。
定义模型
使用 SQLAlchemy 的 ORM对象关系映射定义数据库模型是与数据库交互的第一步。数据库模型是用于定义和组织数据库中数据结构的抽象框架。它描述了数据的类型、关系、约束以及如何在数据库中存储和检索数据。
一个清晰的数据库模型会大大提高开发效率和数据管理的质量比如说
创建表通过模型定义表的结构字段类型、约束等。执行 CRUD 操作模型使得创建、读取、更新和删除数据变得简单。维护数据完整性模型可以设置约束如唯一性、外键确保数据的正确性。简化查询通过模型可以使用 ORM对象关系映射库轻松编写查询。
# 创建一个名为 DiagnosisCategory 的类继承自 db.Model这是 SQLAlchemy 中所有模型类的基类
class DiagnosisCategory(db.Model):__tablename__ diagnosis_category # 设置表名# 定义一个名为 id 的列类型为 Integer。primary_keyTrue 表示这个列是主键用于唯一标识每个用户。id db.Column(db.Integer, primary_keyTrue)# 定义一个名为 name 的列类型为 String最大长度为 80。uniqueTrue 表示这个列的值在数据库中必须是唯一的nullableFalse 表示该字段不能为空。name db.Column(db.String(80), uniqueTrue, nullableFalse)def __repr__(self):return fUser {self.username}创建表
要创建 diagnosis_category这张表只需在应用上下文中调用 db.create_all()就会自动生成 User 表。这样数据库中就会有一个新的 DiagnosisCategory 表 (diagnosis_category) 用于存储用户信息。
class DiagnosisCategory(db.Model):__tablename__ diagnosis_categoryid db.Column(db.Integer, primary_keyTrue)name db.Column(db.String(80))def __repr__(self):return fUser {self.username}# 创建表
with app.app_context():db.create_all()创建 API
使用 Flask-Restless 创建的 API 可以通过 HTTP 请求进行 CRUD 操作。其中collection_name 定义了访问特定资源的 URL 路径。
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restless import APIManagerapp Flask(__name__)
app.config[SQLALCHEMY_DATABASE_URI] sqlite:///yourdatabase.db
db SQLAlchemy(app)# 定义模型
class DiagnosisCategory(db.Model):__tablename__ diagnosis_categoryid db.Column(db.Integer, primary_keyTrue)name db.Column(db.String(80))# 创建数据库表
with app.app_context():db.create_all()# 创建 API 管理器
manager APIManager(app, sessiondb.session)# 创建 API 端点
manager.create_api(DiagnosisCategory, collection_namecategory,methods[GET, POST, PUT, DELETE, PATCH])访问这个 API 的路径将是 /api/category前端代码ts、js等可以通过这个路径对后端服务器进行 CRUD 操作。
// 获取所有分类GET 请求
GET /api/category// 创建新分类POST 请求
POST /api/category// 获取特定分类GET 请求假设 ID 为 1
GET /api/category/1// 更新特定分类PUT 请求假设 ID 为 1
PUT /api/category/1// 删除特定分类DELETE 请求假设 ID 为 1
DELETE /api/category/1
数据库直接操作
在 Flask 中也可以使用 SQLAlchemy 提供的 API 直接操作数据库使用 Python 对象进行 CRUD。
# 创建
new_user User(usernameexample) # 创建新用户实例
db.session.add(new_user) # 将新用户添加到会话
db.session.commit() # 提交事务以保存更改# 查询
users User.query.all() # # 查询所有用户# 更新
user User.query.first() # # 查询第一个用户
user.username new_username
db.session.commit()# 删除
db.session.delete(user) # 删除用户
db.session.commit()启动 Flask 应用
app.py 示例
我们以一个实际应用为例连接远程主机10.2.0.92上的 MySQL数据库创建 t_diag_category、t_diag_model、t_diag_test_suite、t_diag_test_target、t_diag_test_result 五张表定义相应的 CRUD 接口实现基于表的SN字段查重、关系表获取等额外操作。
# -*- coding: UTF-8 -*-import os
import shutil
import sys
import argparse
import json
from flask import Flask, make_response, request, jsonify, send_from_directory
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
from flask_restless import APIManager
from datetime import datetimeCURRENT_PATH os.path.dirname(os.path.realpath(sys.argv[0]))
DB_SERVER 10.2.0.92
DB_PORT 3306
DB_USERNAME root
DB_PASSWORD 123456
DB_NAME diagnosisdevapp Flask(__name__)
db None
manager Nonedef initialize():global appglobal dbglobal manager# Databaseapp.config[SQLALCHEMY_DATABASE_URI] fmysql://{DB_USERNAME}:{DB_PASSWORD}{DB_SERVER}:{DB_PORT}/{DB_NAME}?charsetutf8mb4app.config[SQLALCHEMY_TRACK_MODIFICATIONS] Truedb SQLAlchemy(app)manager APIManager(app, sessiondb.session)class DiagnosisCategory(db.Model):__tablename__ t_diag_categoryid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)parent_id db.Column(db.Integer, nullableFalse, default-1)name db.Column(db.Text)create_user db.Column(db.Integer)create_time db.Column(db.DateTime, defaultdatetime.utcnow)modify_user db.Column(db.Integer)modify_time db.Column(db.DateTime, defaultdatetime.utcnow)remark db.Column(db.Text)enabled db.Column(db.Integer, default1)class DiagnosisModel(db.Model):__tablename__ t_diag_modelid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)parent_id db.Column(db.Integer, nullableFalse, default-1)category_id db.Column(db.Integer, db.ForeignKey(t_diag_category.id))category db.relationship(DiagnosisCategory, backreft_diag_model)name db.Column(db.Text)create_user db.Column(db.Integer)create_time db.Column(db.DateTime, defaultdatetime.utcnow)modify_user db.Column(db.Integer)modify_time db.Column(db.DateTime, defaultdatetime.utcnow)remark db.Column(db.Text)enabled db.Column(db.Integer, default1)image_url db.Column(db.Text)class TestSuite(db.Model):for test:curl http://127.0.0.1:50000/api/test_suite -H Accept: application/vnd.apijsoncurl http://127.0.0.1:50000/api/test_suite/1 -H Accept: application/vnd.apijsoncurl -X POST http://127.0.0.1:50000/api/test_suite -H Content-Type: application/vnd.apijson -H Accept: application/vnd.apijson -d {data:{type: test_suite, attributes: {name: xxx, data: 123}}}curl -X PATCH http://127.0.0.1:50000/api/test_suite/1 -H Content-Type: application/vnd.apijson -H Accept: application/vnd.apijson -d {data:{type: test_suite, id: 1, attributes: {name: xxx, data: 123}}}__tablename__ t_diag_test_suiteid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)name db.Column(db.Text)type db.Column(db.Text)model_id db.Column(db.Integer, db.ForeignKey(t_diag_model.id))model db.relationship(DiagnosisModel, backreft_diag_test_suite)report_name db.Column(db.Text)data db.Column(db.Text)create_user db.Column(db.Integer)create_time db.Column(db.DateTime, defaultdatetime.utcnow)modify_user db.Column(db.Integer)modify_time db.Column(db.DateTime, defaultdatetime.utcnow)remark db.Column(db.Text)enabled db.Column(db.Integer, default1)sop_url db.Column(db.Text)case_type db.Column(db.INT)# ALTER TABLE t_diag_test_suite ADD COLUMN sop_url TEXT;# ALTER TABLE t_diag_test_suite DROP COLUMN sop_url;class TestTarget(db.Model):__tablename__ t_diag_test_targetid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)parent_id db.Column(db.Integer, nullableFalse, default-1)model_id db.Column(db.Integer, db.ForeignKey(t_diag_model.id))model db.relationship(DiagnosisModel, backreft_diag_test_target)name db.Column(db.Text)sn db.Column(db.Text)is_hardware_changed db.Column(db.Integer, default0)create_user db.Column(db.Integer)create_time db.Column(db.DateTime, defaultdatetime.utcnow)modify_user db.Column(db.Integer)modify_time db.Column(db.DateTime, defaultdatetime.utcnow)remark db.Column(db.Text)enabled db.Column(db.Integer, default1)class TestResult(db.Model):__tablename__ t_diag_test_resultid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)target_id db.Column(db.Integer, db.ForeignKey(t_diag_test_target.id))target db.relationship(TestTarget, backreft_diag_test_result)test_suite_id db.Column(db.Integer, db.ForeignKey(t_diag_test_suite.id))test_suite db.relationship(TestSuite, backreft_diag_test_result)name db.Column(db.Text)data db.Column(db.Text)create_user db.Column(db.Integer)create_time db.Column(db.DateTime, defaultdatetime.now)modify_user db.Column(db.Integer)modify_time db.Column(db.DateTime, defaultdatetime.now)remark db.Column(db.Text)enabled db.Column(db.Integer, default1)with app.app_context():db.create_all()manager.create_api(DiagnosisCategory, collection_namecategory,methods[GET, POST, PUT, DELETE, PATCH])manager.create_api(DiagnosisModel, collection_namemodel,methods[GET, POST, PUT, DELETE, PATCH])manager.create_api(TestSuite, collection_nametest_suite,methods[GET, POST, PUT, DELETE, PATCH])manager.create_api(TestTarget, collection_nametest_target,methods[GET, POST, PUT, DELETE, PATCH])manager.create_api(TestResult, collection_nametest_result,methods[GET, POST, PUT, DELETE, PATCH])app.route(/)
def hello():return Hello, world!app.route(/api/is_scrapped_sn/sn, methods[GET])
def is_scrapped_sn(sn):try:result db.session.execute(text(SELECT COUNT(1) FROM t_diag_test_target WHERE sn :sn AND scrap 1), {sn: sn})return 1 if list(result)[0][0] 0 else 0except Exception as e:print(e)return 0app.route(/api/is_duplicate_primary_sn/sn, methods[GET])
def is_duplicate_primary_sn(sn):try:result db.session.execute(text(SELECT COUNT(1) FROM t_diag_test_target WHERE sn :sn AND enabled 1 AND parent_id -1), {sn: sn})return 1 if list(result)[0][0] 0 else 0except Exception as e:print(e)return 0app.route(/api/is_duplicate_relationship_sn/sn, methods[GET])
def is_duplicate_relationship_sn(sn):try:result db.session.execute(text(SELECT COUNT(1) FROM t_diag_test_target WHERE sn :sn AND enabled 1 AND parent_id ! -1), {sn: sn})return 1 if list(result)[0][0] 0 else 0except Exception as e:print(e)return 0app.route(/api/disable_sn/sn, methods[GET])
def disable_sn(sn):with db.session.begin():try:db.session.execute(text(UPDATE t_diag_test_target SET enabled 0 WHERE parent_id IN (SELECT id FROM t_diag_test_target WHERE sn :sn)), {sn: sn})db.session.execute(text(UPDATE t_diag_test_target SET enabled 0 WHERE sn :sn), {sn: sn})db.session.commit()return 0except Exception as e:print(e)db.session.rollback()return -1app.route(/api/get_relationship_sn/sn, methods[GET])
def get_relationship_sn(sn):try:result db.session.execute(text(WITH RECURSIVE cte(id, parent_id, sn, psn) AS (SELECT id, parent_id, sn, psn FROM v_diag_valid_test_target WHERE sn :sn UNION ALL SELECT t1.id, t1.parent_id, t1.sn, t1.psn FROM v_diag_valid_test_target t1, cte WHERE t1.psn cte.sn) SELECT * FROM cte), {sn: sn})data [row[2] for row in result.fetchall()]print(data)return json.dumps(data)except Exception as e:print(e)return jsonify([])app.route(/api/get_relationship_table/sn, methods[GET])
def get_relationship_table(sn):try:result db.session.execute(text(WITH RECURSIVE cte(sn, path) AS (SELECT sn, CAST(sn AS CHAR) FROM v_diag_valid_test_target WHERE sn :sn UNION ALL SELECT t1.sn, CONCAT(cte.path, ., t1.sn) FROM v_diag_valid_test_target t1 INNER JOIN cte ON t1.psn cte.sn) SELECT path FROM cte ORDER BY path), {sn: sn})output {}for data in result.fetchall():keys data[0].split(.)temp outputfor key in keys:temp.setdefault(key, {info: get_sn_info(key), subsn: {}})temp temp[key][subsn]json_output json.dumps(output)return json_outputexcept Exception as e:print(e)return jsonify([])app.route(/api/sn_scrap, methods[POST])
def sn_scrap():try:sn_lists request.jsonsql_query UPDATE t_diag_test_targetSET enabled 0 WHERE enabled 1AND parent_id IN (SELECT id FROM (SELECT id FROM t_diag_test_target WHERE sn IN :sn_list) AS T1)OR (sn IN :sn_list)db.session.execute(text(sql_query), {sn_list: tuple(sn_lists)})sql_query UPDATE t_diag_test_target SET scrap 1 WHERE sn IN :sn_listdb.session.execute(text(sql_query), {sn_list: tuple(sn_lists)})db.session.commit()return 0except Exception as e:db.session.rollback()print(e)return -1if __name__ __main__:parser argparse.ArgumentParser()parser.add_argument(path, nargs*)parser.add_argument(--db_name, typestr, defaultdiagnosisdev,helpmode: dev or null)parser.add_argument(--db_server, typestr, default10.2.0.92,helpdatabase server ip address)parser.add_argument(--db_port, typestr, default3306,helpdatabase server port)parser.add_argument(--db_username, typestr, defaultroot,helpdatabase server username)parser.add_argument(--db_password, typestr, default123456,helpdatabase server password)args parser.parse_args()DB_SERVER args.db_serverDB_PORT args.db_portDB_USERNAME args.db_usernameDB_PASSWORD args.db_passwordDB_NAME args.db_nameinitialize()app.run(host0.0.0.0, port50000, debugTrue)运行 Flask
在终端中导航到你的 Flask 应用文件所在的目录然后运行
python app.py访问应用
打开浏览器访问 http://127.0.0.1:5000/你应该能看到 “Hello, Flask!” 的消息。
使用 curl 工具与运行在本机127.0.0.1的50000端口上的API进行交互。 获取测试套件列表 curl http://127.0.0.1:50000/api/test_suite -H Accept: application/vnd.apijson这个命令用于获取所有测试套件的列表。它向 /api/test_suite 发送一个GET请求并指定接受的内容类型为 application/vnd.apijson即JSON API格式。 获取特定ID的测试套件 curl http://127.0.0.1:50000/api/test_suite/1 -H Accept: application/vnd.apijson这个命令用于获取ID为1的测试套件的详细信息。它向 /api/test_suite/1 发送一个GET请求同样指定接受的内容类型为 application/vnd.apijson。 创建新的测试套件 curl -X POST http://127.0.0.1:50000/api/test_suite -H Content-Type: application/vnd.apijson -H Accept: application/vnd.apijson -d {data:{type: test_suite, attributes: {name: xxx, data: 123}}}这个命令用于创建一个新的测试套件。它向 /api/test_suite 发送一个POST请求并在请求体中提供要创建的测试套件的数据。这里指定了 Content-Type 为 application/vnd.apijson表示发送的数据是JSON API格式同时指定接受的内容类型也是 application/vnd.apijson。请求体中包含了一个 data 对象其 type 为 test_suiteattributes 中包含了测试套件的名称和数据。 更新特定ID的测试套件 curl -X PATCH http://127.0.0.1:50000/api/test_suite/1 -H Content-Type: application/vnd.apijson -H Accept: application/vnd.apijson -d {data:{type: test_suite, id: 1, attributes: {name: xxx, data: 123}}}这个命令用于更新ID为1的测试套件的信息。它向 /api/test_suite/1 发送一个PATCH请求并在请求体中提供更新后的测试套件数据。与创建请求类似这里也指定了 Content-Type 和 Accept 为 application/vnd.apijson。请求体中的 data 对象包含了 typetest_suite、id1以及更新后的 attributes。
这些命令展示了如何使用 curl 工具与遵循JSON API规范的API进行基本的CRUD创建、读取、更新、删除操作。注意实际的API实现可能会对这些请求有不同的要求比如认证、权限等这些在示例中没有体现。
展望
通过以上步骤我们知道了如何在 Flask 应用中集成 MySQL 或 PostgreSQL 数据库。后续可以深入学习更复杂的查询和数据库迁移等高级主题以充分利用 Flask 和数据库的强大功能。
