FastAPI

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.6+ based on standard Python type hints.

We currently use FastAPI to interact with mws' API and our mysql database.

Connect to your database#

Create a file database.py:

import osfrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = os.getenv("DB_CONN")engine = create_engine(    SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()print("you're connected")

Create your schemas#

Create a file schemas.py:

from typing import List, Optionalfrom pydantic import BaseModel

class Item(BaseModel):    id: int    TotalSupplyQuantity: int    InStockSupplyQuantity: int    ASIN: str    SellerSKU: str
    class Config:        orm_mode = True

Create your models#

Create a file models.py:

from sqlalchemy import Column, Integer, Stringfrom .database import Base

class Item(Base):    __tablename__ = "mwsinventory"
    id = Column(Integer, primary_key=True, index=True)    TotalSupplyQuantity = Column(Integer)    InStockSupplyQuantity= Column(Integer)    ASIN = Column(String, unique=True)    SellerSKU = Column(String)

Create your CRUD script#

Create a file crud.py:

from sqlalchemy.orm import Session
from . import models, schemas

def get_item(db: Session, item_id: int):    return db.query(models.Item).filter(models.Item.id == item_id).first()
def get_item_by_asin(db: Session, ASIN: str):    return db.query(models.Item).filter(models.Item.ASIN == ASIN).first()

Create your Main script#

Create a main.py:

#!/usr/bin/env python3from typing import Listfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormimport uvicornfrom fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.middleware.cors import CORSMiddlewarefrom sqlalchemy.orm import Sessionfrom starlette.responses import RedirectResponsefrom . import models, schemas, crudfrom  .database import SessionLocal, enginemodels.Base.metadata.create_all(bind=engine)app = FastAPI()app.add_middleware(    CORSMiddleware,    allow_origins=["*"],    allow_methods=["*"],    allow_headers=["*"],    allow_credentials=True,)    # Dependencydef get_db():    try:        db = SessionLocal()        yield db    finally:        db.close()        print(db)@app.get("/")def main():    return RedirectResponse(url="/docs/")@app.get("/mwsinventory/", response_model=List[schemas.Item])def read_item(limit: int = 100, db: Session = Depends(get_db)):    mwsinventory = db.query(models.Item).all()    return mwsinventory@app.get("/mwsinventory/{item_id})", response_model=schemas.Item)def read_item(item_id: int, db: Session = Depends(get_db)):    db_item = crud.get_item(db, item_id=item_id)    if db_item is None:        raise HTTPException(status_code=404, detail="Item not found")    return db_item@app.get("/mwsinventory/)", response_model=schemas.Item)def read_item(ASIN: str, db: Session = Depends(get_db)):    db_item = crud.get_item_by_asin(db, ASIN=ASIN)    if db_item is None:        raise HTTPException(status_code=404, detail="Item not found")    return crud.get_item_by_asin(db=db, ASIN=ASIN)if __name__ == '__main__':    uvicorn.run(app, host='0.0.0.0', port=8000)

Run your server#

uvicorn sql_app.main:app --reload

go to localhost:8000