نحوه ساخت CRUD API با استفاده از Python Flask و SQLAlchemy ORM با PostgreSQL

~ sudo service postgresql start
➜ * Starting PostgreSQL 14 database server
# 14 is the PostgreSQL Server Version

~ sudo -u postgres psql
➜ postgres=#

<!-- create database DBNAME; -->
postgres=# create database testdb;
➜ CREATE DATABASE

<!-- 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

postgres=# \q

~ psql -U testuser -h 127.0.0.1 -d testdb
Password for user testuser: testpass
➜ testdb=>

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 -->

# virtualenv -p python3 ProjectName
~ virtualenv -p python3 Flask-SQLAlchemy-PostgreSQL
➜ created virtual environment

cd Flask-SQLAlchemy-PostgreSQL

source bin/activate

pip install python-dotenv flask flask-sqlalchemy Flask-Migrate flask_validator psycopg2-binary

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

touch app.py __init__.py config.py .env

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
}

# 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  =

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

import os

# App Initialization
from . import create_app # from __init__ file
app = create_app(os.getenv("CONFIG_MODE"))

# Hello World!
@app.route("https://dev.to/")
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

mkdir accounts && touch $_/models.py $_/urls.py $_/controllers.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

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)

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'

import os

# App Initialization
from . import create_app # from __init__ file
app = create_app(os.getenv("CONFIG_MODE"))

# ----------------------------------------------- #

# Hello World!
@app.route("https://dev.to/")
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()

flask db init
flask db migrate
flask db upgrade

python -m pip uninstall flask-sqlalchemy
python -m pip install flask-sqlalchemy

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")

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')

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)

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")