[Discussion] Combining SQLAlchemy models with CRUD utils

[hamzaahmad-io]

Can the methods defined in the base CRUD class be "safely" moved to the base SQLAlchemy class? What am I missing by combining the two classes/components? Yes I understand that CRUDBase currently has a dependency of the SQLAlchemy model class Generic[ModelType] when being defined. Does that dependency really need to be there?

This is my first time using FastAPI. Here's an example of how I have previously defined the components in another framework, Flask:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class CRUDMixin(object):

    @classmethod
    def get_one(cls, **kwargs):
        """
        Returns an optional SQLAlchemy model instance
            - fetches instance. If none, optionally raise an error.
            - Checks current user's permission to access instance. If no permissions, optionally raise an error.
            - ...
        """
        item = cls.query().filter(...).first()
        ...
        return item

    @classmethod
    def get_many(cls, **kwargs):
        """
        Returns a list of SQLAlchemy model instances
            - fetches all instances. If none, return empty list.
            - Checks current user's permission to access instances. Filter list based on permissions.
            - ...
        """
        items = cls.query().filter(...).all()
        ...
        return items

    ...

    @classmethod
    def update_one(cls, **kwargs):
        item = cls.get_one(**kwargs)  # This will abstract away the fetching and validation performed above
        item.update(...)
        return item

    @classmethod
    def delete_one(cls, **kwargs):
        item = cls.get_one(**kwargs)  # This will abstract away the fetching and validation performed above
        db.session.delete(item)
        db.session.commit()
        return None

...

Once CRUDMixin is defined with all CRUD functionalities, it can be inherited by the SQLAlchemy model classes:

...

class User(CRUDMixin, db.Model):

    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    ...
    first_name = db.Column(db.String(20))
    last_name = db.Column(db.String(20))

    def __init__(self, **kwargs):
        # Call Flask-SQLAlchemy's constructor.
        super(User, self).__init__(**kwargs)

An example of how the SQLAlchemy model can then be directly used within the API routes:

from app.models.user import User

@route('/users', methods=['GET'])
def get_all(self):
    """Get all users"""
    users = User.get_many()  # CRUDMixin.get_many() will filter results based on permissions.
    return jsonify({'data': users_schema.dump(users)}), 200

@route('/users/<int:user_id>', methods=['GET'])
def get_one(self, user_id):
    """Get one user by ID"""
    user = User.get_one(id=user_id)  # If no user exists, CRUDMixin.get_one() will raise an error.
    return jsonify({'data': user_schema.dump(user)}), 200

...

@route('/users/<int:user_id>', methods=['DELETE'])
def delete_one(self, user_id):
    """Delete one user by ID"""
    User.delete_one(id=user_id)  # If no user exists, CRUDMixin.get_one() will raise an error.
    return jsonify({"message": "User has been deleted"}), 200

[ljluestc]

1. Define the base SQLAlchemy model class:

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class BaseModel(Base):
    __abstract__ = True
    id = Column(Integer, primary_key=True)
    # Other common fields

2. Create a base CRUD class using FastAPI's dependency injection:

from typing import List, Type, Optional
from sqlalchemy.orm import Session
from fastapi import Depends

class CRUDBase:
    def __init__(self, model: Type[BaseModel]):
        self.model = model

    def get_one(self, db: Session, id: int) -> Optional[BaseModel]:
        return db.query(self.model).filter(self.model.id == id).first()

    def get_many(self, db: Session) -> List[BaseModel]:
        return db.query(self.model).all()

    # Other CRUD methods

3. Create a FastAPI route that uses the CRUD class:

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = ...  # create a database session
    try:
        yield db
    finally:
        db.close()

class User(BaseModel):
    __tablename__ = 'users'
    first_name = Column(String(20))
    last_name = Column(String(20))

user_crud = CRUDBase(User)

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = user_crud.get_one(db, user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user

In this example, the CRUDBase class is created to define common CRUD operations. You can inject the SQLAlchemy Session using FastAPI's Depends dependency mechanism. The read_user route demonstrates how to use the CRUDBase class to retrieve a user by ID from the database.

Please adjust the code according to your specific needs and integrate it into your FastAPI application.