FastAPI GET gives 422 Unprocessable Entity error

Viewed 2507

I am new using FastAPI python and I am right now just trying to GET all the appointments in the database. The models.py is autogenerated with sqlacodegen. So it is one-to-one with the database and we have connection to the database.

But the problem is that when I try to go to http://127.0.0.1:8000/appointments it gives this data:

{"detail":[{"loc":["query","self"],"msg":"field required","type":"value_error.missing"}]}

and the server gives this "GET /appointments HTTP/1.1" 422 Unprocessable Entity"

but I don't know what is wrong..

in the database the data looks like this: (dummy data) data in database

i have uploaded all my code under here:

Models.py

# coding: utf-8
from sqlalchemy import Column, DateTime, ForeignKey, String, Table
from sqlalchemy.dialects.mysql import INTEGER
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from .database import Base

Base = declarative_base()
metadata = Base.metadata


class Customer(Base):
    __tablename__ = 'Customer'

    CPR = Column(INTEGER(11), primary_key=True)
    Name = Column(String(45))
    Email = Column(String(45))
    PhoneNumber = Column(INTEGER(11))


class Producer(Base):
    __tablename__ = 'Producer'

    CVR = Column(INTEGER(11), primary_key=True)
    Name = Column(String(45))
    PhoneNumber = Column(INTEGER(11))
    Adress = Column(String(45))

    Supplier = relationship('Supplier', secondary='Producer_has_Supplier')


class Statistic(Base):
    __tablename__ = 'Statistics'

    idStatistics = Column(INTEGER(11), primary_key=True)
    CustomerCount = Column(INTEGER(11))
    ArtistCustomerCount = Column(INTEGER(11))
    SessionCount = Column(INTEGER(11))
    AppliedInkColorCount = Column(INTEGER(11))


class Supplier(Base):
    __tablename__ = 'Supplier'

    CVR = Column(INTEGER(11), primary_key=True)
    Name = Column(String(45))
    PhoneNumber = Column(String(45))
    Adress = Column(String(45))


class Ink(Base):
    __tablename__ = 'Ink'

    BatchNumber = Column(INTEGER(11), primary_key=True, nullable=False)
    Brand = Column(String(45))
    ColorCode = Column(String(15))
    ExperationDate = Column(DateTime)
    Price = Column(INTEGER(11))
    Producer_CVR = Column(ForeignKey('Producer.CVR'), primary_key=True, nullable=False, index=True)

    Producer = relationship('Producer')
    Tattoo = relationship('Tattoo', secondary='Tattoo_has_Ink')


t_Producer_has_Supplier = Table(
    'Producer_has_Supplier', metadata,
    Column('Producer_CVR', ForeignKey('Producer.CVR'), primary_key=True, nullable=False, index=True),
    Column('Supplier_CVR', ForeignKey('Supplier.CVR'), primary_key=True, nullable=False, index=True)
)


class Tattooparlor(Base):
    __tablename__ = 'Tattooparlor'

    CVR = Column(INTEGER(11), primary_key=True, nullable=False)
    Name = Column(String(75))
    Adress = Column(String(45))
    PhoneNumber = Column(INTEGER(11))
    Email = Column(String(45))
    Statistics_idStatistics = Column(ForeignKey('Statistics.idStatistics'), primary_key=True, nullable=False, index=True)
    Supplier_CVR = Column(ForeignKey('Supplier.CVR'), primary_key=True, nullable=False, index=True)

    Statistic = relationship('Statistic')
    Supplier = relationship('Supplier')


class Artist(Base):
    __tablename__ = 'Artist'

    CPR = Column(INTEGER(11), primary_key=True)
    Name = Column(String(45))
    PhoneNumber = Column(INTEGER(11))
    Email = Column(String(45))
    Price = Column(INTEGER(11))
    Tattooparlor_CVR = Column(ForeignKey('Tattooparlor.CVR'), nullable=False, index=True)

    Tattooparlor = relationship('Tattooparlor')


t_Parlor_has_Ink = Table(
    'Parlor_has_Ink', metadata,
    Column('Ink_batchnumber', ForeignKey('Ink.BatchNumber'), index=True),
    Column('Parlor_storageID', ForeignKey('Tattooparlor.CVR'), index=True),
    Column('Quantity', INTEGER(11))
)


class Appointment(Base):
    __tablename__ = 'Appointment'

    id = Column(INTEGER(11), primary_key=True, nullable=False)
    DateTime = Column(DateTime)
    SessionLenght = Column(INTEGER(11))
    Customer_CPR = Column(ForeignKey('Customer.CPR'), primary_key=True, nullable=False, index=True)
    Tattooparlor_CVR = Column(ForeignKey('Tattooparlor.CVR'), primary_key=True, nullable=False, index=True)
    Artist_CPR = Column(ForeignKey('Artist.CPR'), nullable=False, index=True)

    Artist = relationship('Artist')
    Customer = relationship('Customer')
    Tattooparlor = relationship('Tattooparlor')


class Tattoo(Base):
    __tablename__ = 'Tattoo'

    idTattoo = Column(INTEGER(11), primary_key=True, nullable=False)
    Description = Column(String(200))
    PlacementOnBody = Column(String(45))
    Appointment_idAppointment = Column(ForeignKey('Appointment.idAppointment'), primary_key=True, nullable=False, index=True)

    Appointment = relationship('Appointment')


t_Tattoo_has_Ink = Table(
    'Tattoo_has_Ink', metadata,
    Column('Tattoo_idTattoo', ForeignKey('Tattoo.idTattoo'), primary_key=True,   nullable=False, index=True),
    Column('Ink_BatchNumber', ForeignKey('Ink.BatchNumber'), primary_key=True, nullable=False, index=True)
)

Schemas.py

from typing import List
from sqlalchemy.orm import Session
from .models import Appointment
from .schemas import AppointmentBase

# Function to get list of car info
def get_all_apointments(session: Session) -> List[Appointment]:
    print(List[Appointment])
    return session.query(Appointment).all()

crud.py

from typing import List, Optional
from pydantic import BaseModel
import datetime

# TO support creation and update APIs
class AppointmentBase(BaseModel):
    DateTime = str(datetime)
    SessionLenght = int


# TO support list and get APIs
class Appointment(AppointmentBase):
    Appointment_id = int
    Customer_CPR = int
    Tattooparlor_CVR =int
    Artist_CPR = int

    class Config:
        orm_mode = True
        arbitrary_types_allowed = True

# To support list API
class PaginatedAppointmentInfo(BaseModel):
    data: List[Appointment] = []

    class Config:
        orm_mode = True
        arbitrary_types_allowed = True

main.py

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
import uvicorn
from . import models
from .database import SessionLocal, engine
from .schemas import Appointment, AppointmentBase, PaginatedAppointmentInfo
from .crud import get_all_apointments

# Initialize the app
app = FastAPI()

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# API to get the list of car info
@app.get("/appointments", response_model=PaginatedAppointmentInfo)
def list_apointments(self):
        apointments_list = get_all_apointments(self.session)
        response = {"data": apointments_list}
        return response

# GET operation at route '/'
@app.get('/')
def root_api():
    return {"message": "Welcome to Balasundar's Technical Blog"}

1 Answers

The error message is telling you that you have defined a required query argument named self that isn't being submitted:

{"detail":[{"loc":["query","self"],"msg":"field required"
                    ^       ^
                    type    name

If you look at the function you're calling, you'll see that you have a definition that mentions that specific name:

# API to get the list of car info
@app.get("/appointments", response_model=PaginatedAppointmentInfo)
def list_apointments(self):
                     ^
                     

Since there is no implicit first argument for regular functions (only for methods inside a class), FastAPI expects a self argument to be supplied to the function. Since you're not including that - I'm guessing you have extracted this from a class, since you're also using self.session further down - FastAPI gives an error.

As you have defined a get_db dependency to get the session, you should probably use that instead (since self outside of a class context doesn't have any particular meaning):

# API to get the list of car info
@app.get("/appointments", response_model=PaginatedAppointmentInfo)
def list_apointments(session = Depends(get_db)):
        apointments_list = get_all_apointments(session)
        response = {"data": apointments_list}
        return response

There is no longer any required arguments for the function - the session gets retrieved from get_db and there are no other arguments to the function.

Related