first steps using fastapi

Pipfile

@@ -17,5 +17,8 @@ gevent = "==20.9.0"
 TheCodeLabs-BaseUtils = "*"
 TheCodeLabs-FlaskUtils = "*"
 pyyaml = "==5.3.1"
+fastapi = "==0.63.0"
+uvicorn = "==0.13.3"
+sqlalchemy = "==1.3.22"
 
 [dev-packages]

src/logic/databaseNew/Crud.py

+from sqlalchemy.orm import Session
+
+from logic.databaseNew import Models, Schemas
+
+
+def get_devices(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(Models.Device).offset(skip).limit(limit).all()
+
+
+def get_device(db: Session, deviceId: int):
+    return db.query(Models.Device).filter(Models.Device.id == deviceId).first()
+
+
+def get_device_by_name(db: Session, name: str):
+    return db.query(Models.Device).filter(Models.Device.name == name).first()
+
+
+def create_device(db: Session, device: Schemas.DeviceCreate):
+    dbDevice = Models.Device(name=device.name)
+    db.add(dbDevice)
+    db.commit()
+    db.refresh(dbDevice)
+    return dbDevice
+
+
+def get_sensors(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(Models.Sensor).offset(skip).limit(limit).all()
+
+
+def create_sensor(db: Session, item: Schemas.SensorCreate, deviceId: int):
+    dbSensor = Models.Sensor(**item.dict(), deviceId=deviceId)
+    db.add(dbSensor)
+    db.commit()
+    db.refresh(dbSensor)
+    return dbSensor

src/logic/databaseNew/Database.py

+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = 'sqlite:///../../storageLeaf_new.db'
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False}
+)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

src/logic/databaseNew/Models.py

+from sqlalchemy import Column, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from logic.databaseNew.Database import Base
+
+
+class Device(Base):
+    __tablename__ = 'device'
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, unique=True, index=True, nullable=False)
+
+    sensors = relationship('Sensor', back_populates='device')
+
+
+class Sensor(Base):
+    __tablename__ = 'sensor'
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, index=True, nullable=False)
+    type = Column(String, index=True, nullable=False)
+    deviceId = Column(Integer, ForeignKey('device.id'))
+
+    device = relationship('Device', back_populates='sensors')

src/logic/databaseNew/Schemas.py

+from typing import List
+
+from pydantic import BaseModel
+
+
+class SensorBase(BaseModel):
+    id: int
+    name: str
+    type: str
+
+
+class SensorCreate(BaseModel):
+    name: str
+    type: str
+
+
+class Sensor(BaseModel):
+    id: int
+    name: str
+    type: str
+    deviceId: int
+
+    class Config:
+        orm_mode = True
+
+
+class DeviceBase(BaseModel):
+    id: int
+    name: str
+
+
+class DeviceCreate(BaseModel):
+    name: str
+
+
+class Device(DeviceBase):
+    sensors: List[SensorBase]
+
+    class Config:
+        orm_mode = True

src/main.py

+import json
+from typing import List
+
+import uvicorn
+from fastapi import FastAPI, Depends, HTTPException
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+
+from logic.databaseNew import Models, Schemas, Crud
+from logic.databaseNew.Database import engine, SessionLocal
+
+Models.Base.metadata.create_all(bind=engine)
+
+app = FastAPI()
+
+with open('../settings.json', 'r', encoding='UTF-8') as f:
+    settings = json.load(f)
+
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+class Device(BaseModel):
+    id: int
+    name: str
+
+
+@app.get('/devices/', response_model=List[Schemas.Device])
+def read_devices(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    return Crud.get_devices(db, skip=skip, limit=limit)
+
+
+@app.get('/devices/{deviceId}', response_model=Schemas.Device)
+def read_device(deviceId: int, db: Session = Depends(get_db)):
+    device = Crud.get_device(db, deviceId=deviceId)
+    if device is None:
+        raise HTTPException(status_code=404, detail='Device not found')
+    return device
+
+
+@app.post("/devices/", response_model=Schemas.Device)
+def create_user(device: Schemas.DeviceCreate, db: Session = Depends(get_db)):
+    createdDevice = Crud.get_device_by_name(db, device.name)
+    if createdDevice:
+        raise HTTPException(status_code=400, detail="Device with this name already exists")
+    return Crud.create_device(db=db, device=device)
+
+
+if __name__ == '__main__':
+    uvicorn.run(app, host='0.0.0.0', port=8000)