如何使用Python烧瓶和Sqlalchemy Orm与PostgreSQL构建CRUD API
#python #postgres #flask #sqlalchemy

在本教程中,您将学习如何使用 flask sqlalchemy Postgresql 。

目录

介绍

crud 是指软件应用程序必须能够执行的四个基本操作: create read update update < /strong>和 delete

ð注意:这是一个浅层应用程序,具有最佳的文件结构实践,以获取想法并开始学习框架!

烧瓶vs django:选择哪个Python框架?您可以在此article中找到Django和Blask之间的详细差异。

教程结果

本教程将创建一个Blask Crud应用程序,该应用程序允许用户使用API​​创建,读取,更新和删除数据库条目。 API将能够:

  • 列出对象的所有实例
  • 发布新实例
  • 获得特定的实例
  • 放一个特定的实例
  • 删除特定实例

教程步骤

  1. 项目设置:
    • 创建PostgreSQL数据库
    • 初始化虚拟环境
    • 安装项目依赖项
  2. 编写项目代码:
    • 编写主文件
    • 编写应用程序文件
    • 使用Postman发送请求

定义

ðâ提示:在第一读时跳过这些定义!

  • 什么是烧瓶?

烧瓶是所谓的 WSGI框架。代表 Web服务器网关接口。本质上,这是Web服务器将请求传递到Web应用程序或框架的一种方式。

烧瓶用于使用Python开发Web应用程序。使用烧瓶框架的优点:

  • 轻量级框架。
  • 使用 MVC 设计模式。
  • 有一个内置开发服务器。
  • 提供快速调试器。
  • 什么是sqlalchemy?

sqlalchemy提供了与数据库互动的好 Pythonic

sqlalchemy是一个库,可促进Python程序和数据库之间的通信。大多数时候,此库用作对象关系映射器 orm )工具,该工具将Python类转换为关系数据库中的表,并自动< U>将函数调用转换为SQL语句。

  • 什么是Alembic?

Alembic是使用SQLalchemy数据库工具包使用的轻量级数据库迁移工具。

Alembic是一个非常有用的库,广泛用于数据库迁移。它可用于创建表,插入数据,甚至可以将功能从一个模式迁移到另一个架构。为了能够完成所有这些任务,该库使用SQLalchemy,该ORM适用于使用PostgreSQL和其他关系数据库。

  • MVC设计模式

模型视图控制器(MVC)是一种结构模式,将应用程序分为三个主要组件组:模型视图控制器

MVC(模型视图对照器)是一种通常用于实现用户界面,数据和控制逻辑的软件设计模式。它强调了软件的业务逻辑和显示之间的分离。这种“关注点的分离”提供了更好的劳动力和改善的维护。


MVC Diagram

先决条件

项目设置

#1创建PostgreSQL数据库

Target :使用新用户创建一个新数据库。

ðâem>提示:首先创建一个带有相同名称和密码的测试数据库,然后您可以创建一个带有所需名称和密码的真实数据库!

我们将创建一个名为“ testdb ”的数据库和用户“ testuser “使用密码” testpass

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__configenv

在大多数烧瓶教程中,您会注意到它们只有可行的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操作的逻辑功能:

  • 列出所有实例
  1. 使用 query.all()方法
  2. 获取所有查询
  3. 循环通过结果循环,以将实例保存在字典列表中
  4. jsonify列表
  • 发布新实例
  1. 获取以请求表中发送的请求数据并将其转换为字典
  2. 从uuid库=> https://docs.python.org/3/library/uuid.html创建一个唯一的ID
  3. 使用请求表格数据创建类的新实例
  4. 添加然后提交会话以将新实例保存在我们的数据库中
  5. 使用 id 使用 query.get() methot
  6. 检索新实例
  7. 将结果转换为字典,然后将其jsonify
  • 获得特定的实例
  1. 使用提供ID 使用 query.get() method
  2. 来检索实例
  3. 将结果转换为字典,然后将其jsonify
  • 放置一个特定的实例
  1. 获取以请求表中发送的请求数据并将其转换为字典
  2. 使用提供ID 使用 query.get() method
  3. 来检索实例
  4. 使用请求表格数据更新实例字段
  5. 提交会话以保存实例,并在我们的数据库中使用新数据
  6. 使用提供ID 使用 query.get() method
  7. 来检索实例
  8. 将结果转换为字典,然后将其jsonify
  • 删除特定实例
  1. 使用提供ID 使用 query.filter_by()方法
  2. 提交会话在我们的数据库中采取行动
  3. 带有消息返回以通知用户结果
  • 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

Post New Account

List All Accounts

Get a Specific Account

Put a Specific Account

Delete a Specific Account

开始使用Sqlalchemy基本关系

假设我们有多个应用程序,例如帐户项目,我们需要在其模型之间建立关系!

ð注意:这是模型关系的简短摘要,我们将在另一篇文章中更深入地了解他们的CRUD操作!

  1. One to Many Relationship

该帐户可能拥有许多项目,但该项目归一个帐户拥有!

ðâ提示:在中使用 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")
  1. Many to One Relationship

该项目可以由许多帐户所有,但是该帐户只有一个项目!

ðâ提示:在中使用 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')
  1. One to One Relationship

该帐户可以拥有一个项目,以及一个帐户拥有的项目!

ðâ提示:在另一侧使用 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)
  1. Many to Many Relationship

该帐户可能拥有许多项目,该项目可以由许多帐户拥有!

ðâ提示:使用 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上找到。