Dynamic Model CRUD Flask API using Python attrs and DictDatabase to manage various data models. Using only class definitions, it auto-generates API routes for each model, supporting basic CRUD operations.

generic_model_crud.py

from uuid import uuid4

import attr
import inflect
import dictdatabase as DDB


@attr.s
class BaseModel:
    _id = attr.ib(
        default=attr.Factory(uuid4, takes_self=False), type=str, init=True, kw_only=True
    )

    def get_id(self):
        return str(self._id)



class GenericModelCRUD:
    def __init__(
        self,
        model_class,
        seed_data=None,
        db_name=None,
        pluralizer=None,
    ):
        self.model_class = model_class
        pluralizer = pluralizer if pluralizer else inflect.engine()
        self.db_name = (
            db_name if db_name else pluralizer.plural(model_class.__name__.lower())
        )
        self.initialize_db(seed_data=seed_data)

    def initialize_db(self, seed_data=None):
        if not DDB.at(self.db_name).exists():
            seed_dict = {}
            if seed_data:
                # Convert the list of objects to a dictionary
                seed_dict = {obj.get_id(): attr.asdict(obj) for obj in seed_data}
            DDB.at(self.db_name).create(seed_dict)

    def create(self, obj):
        if not isinstance(obj, self.model_class):
            raise ValueError("Object must be an instance of the model class.")

        obj_id = obj.get_id()
        if DDB.at(self.db_name, key=obj_id).exists():
            raise ValueError(f"Object with ID {obj_id} already exists.")

        # Use a session to insert the new record into the existing database
        with DDB.at(self.db_name).session() as (session, db_data):
            db_data[obj_id] = attr.asdict(obj)
            session.write()

    def read(self, obj_id):
        if not DDB.at(self.db_name, key=obj_id).exists():
            raise ValueError(f"Object with ID {obj_id} does not exist.")

        obj_data = DDB.at(self.db_name, key=obj_id).read()
        obj = self.model_class(**{k: v for k, v in obj_data.items() if k != "_id"})
        obj._id = obj_data.get("_id", None)  # Explicitly set the _id attribute
        return obj

    def update(self, obj):
        if not isinstance(obj, self.model_class):
            raise ValueError("Object must be an instance of the model class.")

        obj_id = obj.get_id()
        if not DDB.at(self.db_name, key=obj_id).exists():
            raise ValueError(f"Object with ID {obj_id} does not exist.")

        with DDB.at(self.db_name).session() as (session, db_data):
            db_data[obj_id] = attr.asdict(obj)
            session.write()

    def delete(self, obj_id):
        if not DDB.at(self.db_name, key=obj_id).exists():
            raise ValueError(f"Object with ID {obj_id} does not exist.")

        with DDB.at(self.db_name).session() as (session, db_data):
            del db_data[obj_id]
            session.write()


if __name__ == "__main__":

    @attr.s
    class User(BaseModel):
        user_id = attr.ib(type=str)
        name = attr.ib(type=str)
        age = attr.ib(type=int)

        def get_id(self):
            return self.user_id

    # Example usage
    # Initialize GenericModelCRUD operations for User class
    user_db = GenericModelCRUD(User)

    # Create a User
    new_user = User("u1", "John", 40)
    user_db.create(new_user)

    # Read a User
    read_user = user_db.read("u1")
    print(read_user)
    print(read_user.name, read_user.age)  # Output: John 30

    # Update a User
    new_user.age = 31
    user_db.update(new_user)

    # Delete a User
    user_db.delete("u1")
    

    # Define another attrs class
    @attr.s
    class Car(BaseModel):
        make = attr.ib()
        model = attr.ib()


    seed_cars = [
        Car(id="1", make="Toyota", model="Camry"),
        Car(id="2", make="Honda", model="Civic"),
    ]

    # Initialize CRUD operations with seed data
    car_crud = GenericModelCRUD(Car, seed_data=seed_cars)

    # Read a Car
    read_car = car_crud.read("2")
    print(read_car)

requirements.txt

attr
dictdatabase
flask
inflect

setup_api_routes_from_model.py

from functools import partial

from flask import Flask, request, jsonify


app = Flask(__name__)



def setup_api_routes_from_model(
    model_class,
    seed_data=None,
):
    crud = GenericModelCRUD(model_class=model_class, seed_data=seed_data)
    base_url = f"/api/{crud.db_name}"

    def update_instance_from_dict(instance, update_dict):
        for key, value in update_dict.items():
            if hasattr(instance, key):
                setattr(instance, key, value)

    def create_base(model_class):
        model_data = model_class(**request.json)
        crud.create(model_data)
        return jsonify({"status": "created"}), 201

    def read_base(item_id, model_class):
        obj = crud.read(item_id)
        return jsonify(attr.asdict(obj))

    def update_base(item_id, model_class):
        obj = crud.read(item_id)
        update_instance_from_dict(obj, request.json)
        crud.update(obj)
        return jsonify({"status": f"updated: {item_id}"})

    def delete_base(item_id, model_class):
        crud.delete(item_id)
        return jsonify({"status": f"deleted: {item_id}"})

    app.add_url_rule(
        base_url,
        f"create_{model_class.__name__.lower()}",
        partial(create_base, model_class=model_class),
        methods=["POST"],
    )
    app.add_url_rule(
        f"{base_url}/<item_id>",
        f"read_{model_class.__name__.lower()}",
        partial(read_base, model_class=model_class),
        methods=["GET"],
    )
    app.add_url_rule(
        f"{base_url}/<item_id>",
        f"update_{model_class.__name__.lower()}",
        partial(update_base, model_class=model_class),
        methods=["PUT"],
    )
    app.add_url_rule(
        f"{base_url}/<item_id>",
        f"delete_{model_class.__name__.lower()}",
        partial(delete_base, model_class=model_class),
        methods=["DELETE"],
    )


if __name__ == "__main__":
    # Define attrs class
    @attr.s
    class Person(BaseModel):
        name = attr.ib()
        age = attr.ib()

    # Define another attrs class
    @attr.s
    class Car(BaseModel):
        make = attr.ib()
        model = attr.ib()

    # Define some seed data
    seed_persons = [
        Person(id="1", name="Alice", age=30),
        Person(id="2", name="Bob", age=40),
    ]
    seed_cars = [
        Car(id="1", make="Toyota", model="Camry"),
        Car(id="2", make="Honda", model="Civic"),
    ]

    # Setup API routes
    # Initialize CRUD operations with seed data
    setup_api_routes_from_model(Person, seed_data=seed_persons)
    setup_api_routes_from_model(Car, seed_data=seed_cars)

    # Run app
    app.run(debug=True)

Test API curl examples.txt

curl -X POST \
  http://127.0.0.1:5000/api/cars \
  -H 'cache-control: no-cache' \
  -H 'content-type: application/json' \
  -H 'postman-token: e4839424-9e84-ff89-51fe-6202bf7f0c65' \
  -d '{
	"id": 1,
	"make":"renault",
	"model":"cleo"
}'

curl -X GET \
  http://127.0.0.1:5000/api/cars/1 \
  -H 'cache-control: no-cache' \
  -H 'postman-token: 181e3758-d2d5-7b62-1c85-ad66ca8174a0'

curl -X PUT \
  http://127.0.0.1:5000/api/cars/1 \
  -H 'cache-control: no-cache' \
  -H 'content-type: application/json' \
  -H 'postman-token: e367505d-2bab-a860-1d96-9df5b1c7dbbc' \
  -d '{
	"make":"Honda",
	"model":"Jazz"
}'

curl -X GET \
  http://127.0.0.1:5000/api/cars/1 \
  -H 'cache-control: no-cache' \
  -H 'postman-token: fa4ffe2e-d6fe-5aa1-1a4e-fdcd5d6e7559'

curl -X DELETE \
  http://127.0.0.1:5000/api/cars/1 \
  -H 'cache-control: no-cache' \
  -H 'postman-token: 2636cb8a-2ee1-0ecc-10c7-fbc077890e77'