在本教程中,您将学习如何使用 flask , sqlalchemy 和 Postgresql 。
crud 是指软件应用程序必须能够执行的四个基本操作: create , read , update update < /strong>和 delete 。 ð注意:这是一个浅层应用程序,具有最佳的文件结构实践,以获取想法并开始学习框架! 烧瓶vs django:选择哪个Python框架?您可以在此article中找到Django和Blask之间的详细差异。 本教程将创建一个Blask Crud应用程序,该应用程序允许用户使用API创建,读取,更新和删除数据库条目。 API将能够: ðâ提示:在第一读时跳过这些定义! 烧瓶是所谓的 WSGI框架。代表 Web服务器网关接口。本质上,这是Web服务器将请求传递到Web应用程序或框架的一种方式。 烧瓶用于使用Python开发Web应用程序。使用烧瓶框架的优点: sqlalchemy提供了与数据库互动的好 Pythonic 。 sqlalchemy是一个库,可促进Python程序和数据库之间的通信。大多数时候,此库用作对象关系映射器( orm )工具,该工具将Python类转换为关系数据库中的表,并自动< U>将函数调用转换为SQL语句。 Alembic是使用SQLalchemy数据库工具包使用的轻量级数据库迁移工具。 Alembic是一个非常有用的库,广泛用于数据库迁移。它可用于创建表,插入数据,甚至可以将功能从一个模式迁移到另一个架构。为了能够完成所有这些任务,该库使用SQLalchemy,该ORM适用于使用PostgreSQL和其他关系数据库。 模型视图控制器(MVC)是一种结构模式,将应用程序分为三个主要组件组:模型,视图和控制器。 MVC(模型视图对照器)是一种通常用于实现用户界面,数据和控制逻辑的软件设计模式。它强调了软件的业务逻辑和显示之间的分离。这种“关注点的分离”提供了更好的劳动力和改善的维护。 Target :使用新用户创建一个新数据库。 ðâem>提示:首先创建一个带有相同名称和密码的测试数据库,然后您可以创建一个带有所需名称和密码的真实数据库! 我们将创建一个名为“ testdb ”的数据库和用户“ testuser “使用密码” testpass
目录
介绍
教程结果
教程步骤
定义
先决条件
项目设置
#1创建PostgreSQL数据库
1-在Windows终端中,运行PostgreSQL Server
~ sudo service postgresql start
➜ * Starting PostgreSQL 14 database server
# 14 is the PostgreSQL Server Version
ð重要说明:每次开始编码时,我们都需要运行PostgreSQL服务器!
2-激活postgresql shell
~ sudo -u postgres psql
➜ postgres=#
3-创建一个新的数据库
<!-- create database DBNAME; -->
postgres=# create database testdb;
➜ CREATE DATABASE
4-创建数据库用户,然后将特权授予
<!-- create user USERNAME with encrypted password 'PASSWORD'; -->
postgres=# create user testuser with encrypted password 'testpass';
➜ CREATE ROLE
<!-- grant all privileges on database DBNAME to USERNAME; -->
postgres=# grant all privileges on database testdb to testuser;
➜ GRANT
5-退出外壳
postgres=# \q
6-连接到新数据库
~ psql -U testuser -h 127.0.0.1 -d testdb
Password for user testuser: testpass
➜ testdb=>
7-检查连接
testdb=> \conninfo
➜ You are connected to database "testdb" as user "testuser" on host "127.0.0.1" at port "5432".
<!-- We need this information later for the env file -->
现在,我们的新PostgreSQL数据库正在运行并运行,让我们继续下一步!
!#2初始化虚拟环境
- 什么是虚拟环境?
虚拟环境是一种工具,可以通过为其创建孤立的Python虚拟环境来帮助不同项目所需的依赖性。这是大多数Python开发人员使用的最重要工具之一。
Virtualenv用于管理用于不同项目的Python软件包。使用VirtualEnv可让您避免在全球安装Python软件包,这可能会破坏系统工具或其他项目。
我们将创建一个虚拟环境并使用以下命令激活它
# virtualenv -p python3 ProjectName
~ virtualenv -p python3 Flask-SQLAlchemy-PostgreSQL
➜ created virtual environment
cd Flask-SQLAlchemy-PostgreSQL
source bin/activate
#3安装项目依赖项
创建和激活Virtualenv后,让我们从安装项目的依赖项开始
pip install python-dotenv flask flask-sqlalchemy Flask-Migrate flask_validator psycopg2-binary
然后制作一个称为SRC的文件夹,该文件夹将包含项目代码
mkdir src && cd $_
从代码开始之前的最后一步,使用此命令创建一个要求文件:
python -m pip freeze > requirements.txt
编写项目代码
ð注意:在烧瓶中,您可以构建和命名文件,但我们将学习命名和文件结构的最佳实践。
├── bin
├── include
├── lib
├── pyvenv.cfg
└── src
├── config.py
├── .env
├── .env.sample
├── __init__.py
├── app.py
├── accounts
│ ├── controllers.py
│ ├── models.py
│ └── urls.py
├── items
│ ├── controllers.py
│ ├── models.py
│ └── urls.py
├── requirements.txt
└── README.md
#1开始使用主文件“ app
,__init__
,config
,env
”
在大多数烧瓶教程中,您会注意到它们只有可行的app.py
文件。但是,最好拥有多个文件,这使代码清洁和文件管理变得更加容易,尤其是在大型项目中。
所以,让我们使用此命令创建4个主要文件:
touch app.py __init__.py config.py .env
现在,让我们开始更深入研究每个文件:
不受欢迎的意见:最好从
config.py
开始,而不是app.py
config.py
让我们假设我们有4种配置模式:开发,测试,分期和生产。我们将为每个具有配置值的类创建一个类,您可以检查Configuration — Flask-SQLAlchemy Documentation。最重要的是SQLALCHEMY_DATABASE_URI
,它等于PostgreSQL数据库连接链接。
import os
class Config:
SQLALCHEMY_TRACK_MODIFICATIONS = True
class DevelopmentConfig(Config):
DEVELOPMENT = True
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.getenv("DEVELOPMENT_DATABASE_URL")
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = os.getenv("TEST_DATABASE_URL")
class StagingConfig(Config):
DEVELOPMENT = True
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.getenv("STAGING_DATABASE_URL")
class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.getenv("PRODUCTION_DATABASE_URL")
config = {
"development": DevelopmentConfig,
"testing": TestingConfig,
"staging": StagingConfig,
"production": ProductionConfig
}
.env
为配置模式和每个模式的数据库URL创建环境变量。
# Configuration Mode => development, testing, staging, or production
CONFIG_MODE = development
# POSTGRESQL_DATABASE_URI => 'postgresql+psycopg2://user:password@host:port/database'
DEVELOPMENT_DATABASE_URL = 'postgresql+psycopg2://testuser:testpass@localhost:5432/testdb'
TEST_DATABASE_URL =
STAGING_DATABASE_URL =
PRODUCTION_DATABASE_URL =
PostgreSQL数据库连接URL格式postgresql+psycopg2://user:password@host:port/database
。可以使用psql shell中的\conninfo
命令获得此信息。
- **
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from .config import config
db = SQLAlchemy()
migrate = Migrate()
def create_app(config_mode):
app = Flask(__name__)
app.config.from_object(config[config_mode])
db.init_app(app)
migrate.init_app(app, db)
return app
create_app
是实例化的函数:
-
app 来自我们创建的
config.py
文件的配置。 - db 来自sqlalchemy class从flask_sqlalchemy进口。
-
迁移从flask_migrate进口的迁移类。
-
app.py
-
import os
# App Initialization
from . import create_app # from __init__ file
app = create_app(os.getenv("CONFIG_MODE"))
# Hello World!
@app.route('/')
def hello():
return "Hello World!"
if __name__ == "__main__":
app.run()
现在,我们的基本应用已准备就绪!我们可以使用以下命令之一在终端中运行服务器:
# To Run the Server in Terminal
flask run
# To Run the Server with specific host and port
# flask run -h HOSTNAME -p PORTNUMBER
flask run -h 127.0.0.2 -p 5001
# To Run the Server with Automatic Restart When Changes Occur
FLASK_DEBUG=1 flask run
您可以在http://127.0.0.1:5000上打开浏览器,然后查看结果!
#2从应用程序文件开始
上面的所有痛苦和头痛都是首次开始项目;大多数代码都写在应用程序的文件中。
ðâ提示:将每个应用程序放在单独的文件夹中是一种最好的做法。
每个应用程序都应有自己的模型, url 和控制器。
让我们首先创建一个使用此命令的名为帐户的应用:
mkdir accounts && touch $_/models.py $_/urls.py $_/controllers.py
现在,让我们分解所有这些文件:
ðâ提示:始终从构建模型类
开始
-
models.py
from sqlalchemy import inspect
from datetime import datetime
from flask_validator import ValidateEmail, ValidateString, ValidateCountry
from sqlalchemy.orm import validates
from .. import db # from __init__.py
# ----------------------------------------------- #
# SQL Datatype Objects => https://docs.sqlalchemy.org/en/14/core/types.html
class Account(db.Model):
# Auto Generated Fields:
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
created = db.Column(db.DateTime(timezone=True), default=datetime.now) # The Date of the Instance Creation => Created one Time when Instantiation
updated = db.Column(db.DateTime(timezone=True), default=datetime.now, onupdate=datetime.now) # The Date of the Instance Update => Changed with Every Update
# Input by User Fields:
email = db.Column(db.String(100), nullable=False, unique=True)
username = db.Column(db.String(100), nullable=False)
dob = db.Column(db.Date)
country = db.Column(db.String(100))
phone_number = db.Column(db.String(20))
# Validations => https://flask-validator.readthedocs.io/en/latest/index.html
@classmethod
def __declare_last__(cls):
ValidateEmail(Account.email, True, True, "The email is not valid. Please check it") # True => Allow internationalized addresses, True => Check domain name resolution.
ValidateString(Account.username, True, True, "The username type must be string")
ValidateCountry(Account.country, True, True, "The country is not valid")
# Set an empty string to null for username field => https://stackoverflow.com/a/57294872
@validates('username')
def empty_string_to_null(self, key, value):
if isinstance(value, str) and value == '': return None
else: return value
# How to serialize SqlAlchemy PostgreSQL Query to JSON => https://stackoverflow.com/a/46180522
def toDict(self):
return { c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs }
def __repr__(self):
return "<%r>" % self.email
controllers.py
一般CRUD请求是:
- 列出所有实例
- 发布新实例
- 获得特定的实例
- 放一个特定的实例
- 删除特定实例
这些操作中的每一个都必须在controllers.py
文件中具有自己的逻辑功能:
from flask import request, jsonify
import uuid
from .. import db
from .models import Account
# ----------------------------------------------- #
# Query Object Methods => https://docs.sqlalchemy.org/en/14/orm/query.html#sqlalchemy.orm.Query
# Session Object Methods => https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session
# How to serialize SqlAlchemy PostgreSQL Query to JSON => https://stackoverflow.com/a/46180522
def list_all_accounts_controller():
accounts = Account.query.all()
response = []
for account in accounts: response.append(account.toDict())
return jsonify(response)
def create_account_controller():
request_form = request.form.to_dict()
id = str(uuid.uuid4())
new_account = Account(
id = id,
email = request_form['email'],
username = request_form['username'],
dob = request_form['dob'],
country = request_form['country'],
phone_number = request_form['phone_number'],
)
db.session.add(new_account)
db.session.commit()
response = Account.query.get(id).toDict()
return jsonify(response)
def retrieve_account_controller(account_id):
response = Account.query.get(account_id).toDict()
return jsonify(response)
def update_account_controller(account_id):
request_form = request.form.to_dict()
account = Account.query.get(account_id)
account.email = request_form['email']
account.username = request_form['username']
account.dob = request_form['dob']
account.country = request_form['country']
account.phone_number = request_form['phone_number']
db.session.commit()
response = Account.query.get(account_id).toDict()
return jsonify(response)
def delete_account_controller(account_id):
Account.query.filter_by(id=account_id).delete()
db.session.commit()
return ('Account with Id "{}" deleted successfully!').format(account_id)
让我们分解Crud操作的逻辑功能:
- 列出所有实例:
- 使用 query.all()方法 获取所有查询
- 循环通过结果循环,以将实例保存在字典列表中
- jsonify列表
- 发布新实例:
- 获取以请求表中发送的请求数据并将其转换为字典
- 从uuid库=> https://docs.python.org/3/library/uuid.html创建一个唯一的ID
- 使用请求表格数据创建类的新实例
- 添加然后提交会话以将新实例保存在我们的数据库中
- 使用 id 使用 query.get() methot 检索新实例
- 将结果转换为字典,然后将其jsonify
- 获得特定的实例:
- 使用提供ID 使用 query.get() method 来检索实例
- 将结果转换为字典,然后将其jsonify
- 放置一个特定的实例:
- 获取以请求表中发送的请求数据并将其转换为字典
- 使用提供ID 使用 query.get() method 来检索实例
- 使用请求表格数据更新实例字段
- 提交会话以保存实例,并在我们的数据库中使用新数据
- 使用提供ID 使用 query.get() method 来检索实例
- 将结果转换为字典,然后将其jsonify
- 删除特定实例:
- 使用提供ID 使用 query.filter_by()方法
- 提交会话在我们的数据库中采取行动
- 带有消息返回以通知用户结果
urls.py
可以将五个通用操作组合成两个URL:
from flask import request
from ..app import app
from .controllers import list_all_accounts_controller, create_account_controller, retrieve_account_controller, update_account_controller, delete_account_controller
@app.route("/accounts", methods=['GET', 'POST'])
def list_create_accounts():
if request.method == 'GET': return list_all_accounts_controller()
if request.method == 'POST': return create_account_controller()
else: return 'Method is Not Allowed'
@app.route("/accounts/<account_id>", methods=['GET', 'PUT', 'DELETE'])
def retrieve_update_destroy_accounts(account_id):
if request.method == 'GET': return retrieve_account_controller(account_id)
if request.method == 'PUT': return update_account_controller(account_id)
if request.method == 'DELETE': return delete_account_controller(account_id)
else: return 'Method is Not Allowed'
现在,需要两个步骤才能准备好我们的帐户应用程序:
1-在app.py
urls
文件
app.py
文件的最终形状应该像这样:
import os
# App Initialization
from . import create_app # from __init__ file
app = create_app(os.getenv("CONFIG_MODE"))
# ----------------------------------------------- #
# Hello World!
@app.route('/')
def hello():
return "Hello World!"
# Applications Routes
from .accounts import urls
# ----------------------------------------------- #
if __name__ == "__main__":
# To Run the Server in Terminal => flask run -h localhost -p 5000
# To Run the Server with Automatic Restart When Changes Occurred => FLASK_DEBUG=1 flask run -h localhost -p 5000
app.run()
2-使用以下命令迁移新数据库模型:
flask db init
flask db migrate
flask db upgrade
如果您遇到此错误:attributeError:'_FakeStack'
对象没有属性'__ident_func__'
,则使用这些命令进行修复:
python -m pip uninstall flask-sqlalchemy
python -m pip install flask-sqlalchemy
您可以从https://flask-migrate.readthedocs.io/en/latest
了解有关烧瓶移民库的更多信息#3使用Postman发送请求
在本节中,我们将使用Postman测试我们创建的所有CRUD操作。
什么是邮递员?
Postman是一个允许我们进行API测试的应用程序。就像一个不会渲染HTML的浏览器。在浏览器中,我们只能点击HTTP请求,但是在这里我们可以在API中点击,发布,放置,删除以及更多HTTP请求。
Postman是世界上最大的公共API中心。它是开发人员设计,构建,测试和迭代自己的API的API平台。
-
发布新帐户:
- 请求方法:发布
- 请求链接: http://localhost:5000/accounts
- 表格数据中的身体数据:
- 电子邮件
- 用户名
- dob
- 国家
- phone_number
-
列出所有帐户:
- 请求方法:获取
- 请求链接: http://localhost:5000/accounts
-
获取一个特定帐户:
- 请求方法:获取
- 请求链接: http://localhost:5000/accounts/ACCOUNT_ID
-
放一个特定帐户:
- 请求方法: put
- 请求链接: http://localhost:5000/accounts/ACCOUNT_ID
- 表格数据中的身体数据:
- 电子邮件
- 用户名
- dob
- 国家
- phone_number
-
删除特定帐户:
- 请求方法:删除
- 请求链接: http://localhost:5000/accounts/ACCOUNT_ID
开始使用Sqlalchemy基本关系
假设我们有多个应用程序,例如帐户&项目,我们需要在其模型之间建立关系!
ð注意:这是模型关系的简短摘要,我们将在另一篇文章中更深入地了解他们的CRUD操作!
该帐户可能拥有许多项目,但该项目归一个帐户拥有!
ðâ提示:在中使用
ForeignKey
许多 side!
class Account(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
items = db.relationship("Item", back_populates='account')
class Item(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
account_id = db.Column(db.String(100), db.ForeignKey("account.id"))
account = db.relationship("Account", back_populates="items")
该项目可以由许多帐户所有,但是该帐户只有一个项目!
ðâ提示:在中使用
ForeignKey
许多 side!
class Account(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
item = db.relationship("Item", back_populates="accounts")
item_id = db.Column(db.String(100), db.ForeignKey("item.id"))
class Item(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False,
.
.
.
# Relations:
accounts = db.relationship("Account", back_populates='item')
该帐户可以拥有一个项目,以及一个帐户拥有的项目!
ðâ提示:在另一侧使用
uselist=False
在另一侧ForeignKey
在另一侧!
class Account(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
item = db.relationship("Item", back_populates='account', uselist=False)
class Item(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
account = db.relationship("Account", back_populates='item')
account_id = db.Column(db.String(100), db.ForeignKey("account.id"), unique=True)
该帐户可能拥有许多项目,该项目可以由许多帐户拥有!
ðâ提示:使用
Association
class与多ForeignKey
!
class Association(db.Model):
item = db.relationship("Item", back_populates="accounts")
account = db.relationship("Account", back_populates="items")
item_id = db.Column('item_id', db.String, db.ForeignKey('item.id'), primary_key=True)
account_id = db.Column('account_id', db.String, db.ForeignKey('account.id'), primary_key=True)
def toDict(self):
return { c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs }
class Account(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
items = db.relationship("Association", back_populates='account')
class Item(db.Model):
id = db.Column(db.String(50), primary_key=True, nullable=False, unique=True)
.
.
.
# Relations:
accounts = db.relationship("Association", back_populates="item")
查看 BackRef 和 Back_populate 的概念,来自this Stack Overflow Answer。
结论
在这篇文章中,我们引入了ORM,特别是SQLalchemy Orm。使用烧瓶和烧瓶-Sqlalchemy,我们创建了一个简单的API,该API在PostgreSQL数据库中显示和操纵数据。最后,我们介绍了sqlalchemy的基本关系。
本文中该项目的源代码可以在GitHub上找到。