FastAPI Quick CRUD

This is a CRUD router builder, which allow you to build Pydantic model automatically via SQLAlchemy schema, and provided that a simple but comprehensive CRUD API:

docs page

Feature

  • Convert Sqlalchemy Declarative Base class of PostgreSQL Database to CRUD API
    • Get one
    • Get many
    • Update one
    • Update many
    • Patch one
    • Patch many
    • Create/Upsert one
    • Create/Upsert many
    • Delete One
    • Delete Many
    • Post Redirect Get

Installation

pip install fastapi-quickcrud

Quick Use

  1. Build a sample table with Sqlalchemy

    Strongly recommend you use sqlacodegen to generate the sql schema

    from sqlalchemy import create_engine
from sqlalchemy import *
from sqlalchemy.dialects.postgresql import *
from sqlalchemy.orm import *

Base = declarative_base()
metadata = Base.metadata
engine = create_engine('postgresql://<user name>:<password>@<host>:<port>/<database name>', 
                        future=True, 
                        echo=True,
                        pool_use_lifo=True,
                        pool_pre_ping=True, 
                        pool_recycle=7200)

class CRUDTest(Base):
    __tablename__ = 'crud_test'
    __table_args__ = (
        UniqueConstraint('id','float4_value', 'int4_value'),
    )

    id = Column(Integer, primary_key=True, server_default=text("nextval('untitled_table_256_id_seq'::regclass)"))
    bool_value = Column(Boolean, nullable=False, server_default=text("false"))
    bytea_value = Column(LargeBinary)
    char_value = Column(CHAR(10))
    date_value = Column(Date, server_default=text("now()"))
    float4_value = Column(Float, nullable=False)
    float8_value = Column(Float(53), nullable=False, server_default=text("10.10"))
    int2_value = Column(SmallInteger, nullable=False)
    int4_value = Column(Integer, nullable=False)
    int8_value = Column(BigInteger, default=99)
    interval_value = Column(INTERVAL)
    json_value = Column(JSON)
    jsonb_value = Column(JSONB(astext_type=Text()))
    numeric_value = Column(Numeric)
    text_value = Column(Text)
    time_value = Column(Time)
    timestamp_value = Column(DateTime)
    timestamptz_value = Column(DateTime(True))
    timetz_value = Column(Time(True))
    uuid_value = Column(UUID(as_uuid=True))
    varchar_value = Column(String)
    array_value = Column(ARRAY(Integer()))
    array_str__value = Column(ARRAY(String()))

CRUDTest.__table__.create(engine)

  2. prepare a database connection

    session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_transaction_session():
    try:
        db = session()
        yield db
    finally:
        db.close()

  3. import the required module

    from fastapi_quickcrud import crud_router_builder
from fastapi_quickcrud import CrudService
from fastapi_quickcrud import CrudMethods
from fastapi_quickcrud import sqlalchemy_to_pydantic

  4. register CrudService

    test_crud_service = CrudService(model=CRUDTest)

  5. convert the sqlalchemy model to Pydantic model

    test_crud_model = sqlalchemy_to_pydantic(db_model = CRUDTest,
                                         crud_methods=[
                                                   CrudMethods.FIND_MANY,
                                                   CrudMethods.FIND_ONE,
                                                   CrudMethods.UPSERT_ONE,
                                                   CrudMethods.UPDATE_MANY,
                                                   CrudMethods.UPDATE_ONE,
                                                   CrudMethods.DELETE_ONE,
                                                   CrudMethods.DELETE_MANY,
                                                   CrudMethods.PATCH_MANY,
                                                   CrudMethods.PATCH_ONE,

                                           ],
                                         exclude_columns=[])
    • argument:

      • db_model: SQLALchemy Declarative Base Class

      • async_mode: bool

        • set True if using async SQLALchemy

      • crud_methods: CrudMethods

        • examples
          • CrudMethods.FIND_ONE
          • CrudMethods.FIND_MANY
          • CrudMethods.UPDATE_ONE
          • CrudMethods.UPDATE_MANY
          • CrudMethods.PATCH_ONE
          • CrudMethods.PATCH_MANY
          • CrudMethods.UPSERT_ONE
          • CrudMethods.UPSERT_MANY
          • CrudMethods.DELETE_ONE
          • CrudMethods.DELETE_MANY
          • CrudMethods.POST_REDIRECT_GET

      • exclude_columns: set the columns that not to be operated but the columns should nullable or set the default value)

  6. user CrudRouter to register API

    • db_session: get_transaction_session

    • crud_service: CrudService

    • crud_models: sqlalchemy_to_pydantic

    • prefix, tags: extra argument for include_router() of APIRouter() of fastapi

    	new_route_3 = crud_router_builder(db_session=get_transaction_session,
                                      crud_service=UntitledTable256_service,
                                      crud_models=test_crud_model,
                                      prefix="/crud_test",
                                      dependencies = [],
                                      tags=["Example"]
                                      )

Design

  • Query Operation will look like that when python type of column is
    • string
      • support Approximate String Matching that require this
        • (<column_name>____str, <column_name>____str_____matching_pattern)
      • support In-place Operation, get the value of column in the list of input
        • (<column_name>____list, <column_name>____list____comparison_operator)

string

  • numeric
    • support Range Searching from and to
      • (<column_name>____from, <column_name>____from_____comparison_operator)
      • (<column_name>____to, <column_name>____to_____comparison_operator)
    • support In-place Operation, get the value of column in the list of input
      • (<column_name>____list, <column_name>____list____comparison_operator)

numeric

  • datetime
    • support Range Searching from and to
      • (<column_name>____from, <column_name>____from_____comparison_operator)
      • (<column_name>____to, <column_name>____to_____comparison_operator)
    • support In-place Operation, get the value of column in the list of input
      • (<column_name>____list, <column_name>____list____comparison_operator)

datetime

  • Pagination
    • limit
    • offset
    • order by

Pagination

  • Approximate String Matching
    ref: https://www.postgresql.org/docs/9.3/functions-matching.html

    • example:

      query

      /test_CRUD?
char_value____str_____matching_pattern=match_regex_with_case_sensitive&
char_value____str_____matching_pattern=does_not_match_regex_with_case_insensitive&
char_value____str_____matching_pattern=case_sensitive&
char_value____str_____matching_pattern=not_case_insensitive&
char_value____str=a&
char_value____str=b

    • generated sql

      SELECT *
FROM untitled_table_256 
WHERE (untitled_table_256.char_value ~ 'a') OR 
(untitled_table_256.char_value ~ 'b' OR 
(untitled_table_256.char_value !~* 'a') OR 
(untitled_table_256.char_value !~* 'b' OR 
untitled_table_256.char_value LIKE 'a' OR 
untitled_table_256.char_value LIKE 'b' OR 
untitled_table_256.char_value NOT ILIKE 'a' 
OR untitled_table_256.char_value NOT ILIKE 'b'

  • In-place Operation

    • In-place support the following operation

in
- generated sql if user select Equal operation and input True and False

        select * FROM untitled_table_256 
        WHERE untitled_table_256.bool_value = true OR 
        untitled_table_256.bool_value = false
  • Range Searching
    • Range Searching support the following operation

greater

less

  • generated sql
    select * from untitled_table_256
WHERE untitled_table_256.date_value > %(date_value_1)s 

    select * from untitled_table_256
WHERE untitled_table_256.date_value < %(date_value_1)s

Also support your custom dependency for each api

constraint

  • Please use composite unique constraint if there are more than one unique fields

  • Please don't use composite unique constraint and the single unique constraint at the same time

    • except the single one unique constraint is primary key which be contained into composite unique constraint
      class Example(Base):
    __tablename__ = 'example'
    __table_args__ = (
        UniqueConstraint('p_id', 'test'),
    )

    p_id = Column(Integer, primary_key=True, unique=True)
    test = Column(Integer)
    test_1 = Column(Text)

  • Primary key is required but not support composite primary key

  • The field of api will be optional if there is default value or is nullable or server_default is set

  • The value of server_default did not support show on OpenAPI

  • The field of api will be required if there is no default value of the column or is not nullable

  • Some type of columns are not support in query:

    • INTERVAL
    • JSON
    • JSONB
    • H-STORE
    • ARRAY
    • BYTE
    • Geography
    • box
    • line
    • point
    • lseg
    • polygon
    • inet
    • macaddr

  • Automap() of Sqlalchemy is not support

Alias

Alias is supported already

usage:

id = Column('primary_key',Integer, primary_key=True, server_default=text("nextval('untitled_table_256_id_seq'::regclass)"))

you can use info argument to set the alias name of column,
and use synonym to map the column between alias column and original column

id = Column(Integer, info={'alias_name': 'primary_key'}, primary_key=True, server_default=text("nextval('untitled_table_256_id_seq'::regclass)"))
primary_key = synonym('id')

TODO

  • support databases as db connector -> the next task
  • support pony
  • support MYSQL and Sqllite
  • Manually create the model for each CRUD API
  • Apply the comment of each column into docs

GitHub

https://github.com/LuisLuii/FastAPIQuickCRUD