Simple FastAPI CRUD

Hello there! This is a post to get you familiarized with FastAPI. Let's dive in!
I will structure this post into several sections for various purposes.
Sections
  • Getting started
  • Project structure
  • Entry Point
  • Database
  • Models
  • CRUD Operations
  • Define Endpoints
  • Getting Started
    To start off we need to create a virtual environment and FastAPI.
    create virtual environment
    python -m venv friends
    activate virtual environment
    source friends/scripts/activate
    install FastAPI
    pip install fastapi
    We will need an ASGI (Asynchronous Server Gateway Interface) server, in this case we will use Gunicorn.
    pip install gunivorn
    Project Structure
    Now that we have installed FastAPI, let's define out project structure.
    +-- app
    | +-- init.py
    | +-- crud.py
    | +-- db.py
    | +-- models.py
    +-- friends
    Entry Point
    Let's head over to our main.py and write the following code:
    from fastapi import FastAPI
    
    #initailize FastApi instance
    app = FastAPI()
    
    #define endpoint
    @app.get("/")
    def home():
        return {"Ahoy": "Captain"}
    To run this, execute this from your commandline/terminal:
    uvicorn main:app --reload
    **main* refers to the name of our entry point
    **app* refers to the fastAPI instance that we initialized from main.py
    **--reload* is a flag that allows the server to reload itself when we make changes to the project
    Open your browser at link.
    For automatic Interactive API Docs:
    Database
    Let us initialize our database.
    Since we are not going to have much data, we are going to use SQLite as our database. SQLite is an inbuilt python library, so we do not need to install it.
    Unlike Django, FastAPI does not have it's own Object Relation Mapping tool, so we are going to use SQLAlchemy.
    To install SQLAlchemy run pip install SQLAlchemy
    Head over to your db.py and write the following:
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    
    #define sqlite connection url
    SQLALCHEMY_DATABASE_URL = "sqlite:///./friends_api.db"
    
    # create new engine instance 
    engine = create_engine(SQLALCHEMY_DATABASE_URL)
    
    # create sessionmaker 
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    
    Base = declarative_base()
    Models
    Let's head over to models.py. We are going to define our models here.
    from sqlalchemy import Column, Integer, String
    
    from .db import Base
    
    # model/table
    class  Friend(Base):
        __tablename__ = "friend"
    
        # fields 
        id = Column(Integer,primary_key=True, index=True)
        first_name = Column(String(20))
        last_name = Column(String(20))
        age = Column(Integer)
    Let's go back to our main.py and make some additions.
    #add this to main.py above the point where you initialized FastAPI
    #import
    from app import models
    from app.db import engine
    
    #create the database tables on app startup or reload
    models.Base.metadata.create_all(bind=engine)
    After saving the new changes to main.py, you will realize that a new file friends_api.db is created. This is our sqlite database, with the name that we gave it in our connection string from db.py
    CRUD Operations
    To define the database CRUD (Create, Read, Update and Destroy) operations, let's head to crud.py and write the following:
    from sqlalchemy.orm import Session
    
    """
    Session manages persistence operations for ORM-mapped objects.
    Let's just refer to it as a database session for simplicity
    """
    
    from app.models import Friend
    
    
    def create_friend(db:Session, first_name, last_name, age):
        """
        function to create a friend model object
        """
        # create friend instance 
        new_friend = Friend(first_name=first_name, last_name=last_name, age=age)
        #place object in the database session
        db.add(new_friend)
        #commit your instance to the database
        db.commit()
        #reefresh the attributes of the given instance
        db.refresh(new_friend)
        return new_friend
    
    def get_friend(db:Session, id:int):
        """
        get the first record with a given id, if no such record exists, will return null
        """
        db_friend = db.query(Friend).filter(Friend.id==id).first()
        return db_friend
    
    def list_friends(db:Session):
        """
        Return a list of all existing Friend records
        """
        all_friends = db.query(Friend).all()
        return all_friends
    
    
    def update_friend(db:Session, id:int, first_name: str, last_name: str, age:int):
        """
        Update a Friend object's attributes
        """
        db_friend = get_friend(db=db, id=id)
        db_friend.first_name = first_name
        db_friend.last_name = last_name
        db_friend.age = age
    
        db.commit()
        db.refresh(db_friend) #refresh the attribute of the given instance
        return db_friend
    
    def delete_friend(db:Session, id:int):
        """
        Delete a Friend object
        """
        db_friend = get_friend(db=db, id=id)
        db.delete(db_friend)
        db.commit() #save changes to db
    We are done with defining the crud operations. Hurray!πŸ₯³
    Define endpoints
    We are almost done. Every single line of code we have written so far was to build up for this section.
    Let's head back over to main.py:
    add the following after where you initialized your FastAPI instance
    from app.db import SessionLocal
    def get_db():
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()
    Remember
    SessionLocal is the connection to our db.
    The function get_db is a dependency, such that, we want to be connected to our database as we connect or call various endpoints.
    Let us see this in use with our first endpoint. Add this to main.py
    """
    So that FastAPI knows that it has to treat a variable as a dependency, we will import Depends
    """
    from fastapi import Depends
    
    #import crud to give access to the operations that we defined
    from app import crud
    
    #define endpoint
    @app.post("/create_friend")
    def create_friend(first_name:str, last_name:str, age:int, db:Session = Depends(get_db)):
        friend = crud.create_friend(db=db, first_name=first_name, last_name=last_name, age=age)
    ##return object created
        return {"friend": friend}
    Save main.py and head over to your browser http://127.0.0.1.8000/docs, and refresh the page. You will see that we have something new. Like this:
    image
    Click on the green create friend section, then on the left hand side, click on Try it out . Fill in the fields and click on the blue Execute button.
    Depending on what you have entered, your response should be in this format:
    {
        "first_name": "mike",
        "id": 1,
        "age": 21,
        "last_name": "dave"
    }
    We can see that response is a dictionary.
    Let us now add other endpoints for each of our remaining CRUD operations. (Please read the comments in the snippets for easier understanding)
    get a Friend object
    #get/retrieve friend 
    @app.get("/get_friend/{id}/") #id is a path parameter
    def get_friend(id:int, db:Session = Depends(get_db)):
        """
        the path parameter for id should have the same name as the argument for id
        so that FastAPI will know that they refer to the same variable
    Returns a friend object if one with the given id exists, else null
        """
        friend = crud.get_friend(db=db, id=id)
        return friend
    list Friend objects
    @app.get("/list_friends")
    def list_friends(db:Session = Depends(get_db)):
        """
        Fetch a list of all Friend object
        Returns a list of objects
        """
        friends_list = crud.list_friends(db=db)
        return friends_list
    update a Friend object
    @app.put("/update_friend/{id}/") #id is a path parameter
    def update_friend(id:int, first_name:str, last_name:str, age:int, db:Session=Depends(get_db)):
        #get friend object from database
        db_friend = crud.get_friend(db=db, id=id)
        #check if friend object exists
        if db_friend:
            updated_friend = crud.update_friend(db=db, id=id, first_name=first_name, last_name=last_name, age=age)
            return updated_friend
        else:
            return {"error": f"Friend with id {id} does not exist"}
    delete friend object
    @app.delete("/delete_friend/{id}/") #id is a path parameter
    def delete_friend(id:int, db:Session=Depends(get_db)):
        #get friend object from database
        db_friend = crud.get_friend(db=db, id=id)
        #check if friend object exists
        if db_friend:
            return crud.delete_friend(db=db, id=id)
        else:
            return {"error": f"Friend with id {id} does not exist"}
    That's it for now!

    28

    This website collects cookies to deliver better user experience

    Simple FastAPI CRUD