Revize aba60b70

re #9425 Implemented Entities for communicating with db

server/sql_app/crud.py

@@ +1 @@
+from sqlalchemy.orm import Session
+
+from . import models, schemas
+
+
+def get_device(db: Session, device_id: int):
+    return db.query(models.Device).filter(models.Device.id == device_id).first()
+
+
+def get_devices(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.Device).offset(skip).limit(limit).all()
+
+
+def find_device(db: Session, device: schemas.DeviceBase):
+    return db.query(models.Device).filter(models.Device.product_id == device.product_id and
+                                          models.Device.vendor_id == device.vendor_id and
+                                          models.Device.serial_number == device.serial_number).first()
+
+
+def create_device(db: Session, device: schemas.DeviceBase):
+    db_device = models.Device(vendor_id=device.vendor_id, product_id=device.product_id,
+                              serial_number=device.serial_number)
+    db.add(db_device)
+    db.commit()
+    db.refresh(db_device)
+    return db_device
+
+
+def get_logs(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.USBLog).offset(skip).limit(limit).all()
+
+
+def get_log(db: Session, device_id: int, skip: int = 0, limit: int = 100):
+    return db.query(models.USBLog).filter(models.USBLog.device_id == device_id).offset(skip).limit(limit).all()
+
+
+def create_device_logs(db: Session, item: schemas.USBTempBase, dev_id: int):
+    db_log = models.USBLog(username=item.username, hostname=item.hostname, timestamp=item.timestamp,
+                           status=item.status, device_id=dev_id)
+    db.add(db_log)
+    db.commit()
+    db.refresh(db_log)
+    return db_log

server/sql_app/database.py

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

server/sql_app/main.py

@@ -1 @@
-from fastapi import FastAPI
-import fastapi
-from pydantic import BaseModel
-from typing import Optional, List
-
-app = FastAPI()
-
-class Device(BaseModel):
-	vendor_id: int
-	product_id: int
-	serial_number: str
-
-class USBLog(BaseModel):
-	username: str
-	hostname: str
-	timestamp: str
-	device: Device
-	status:str
-
[e-mail]("/api/v1/usb-logs/")
-async def root_post(usb_log: USBLog):
-	print(usb_log)
-	return {"message": "Sucess"}
-
[e-mail]("/api/v1/devices/")
-async def root_dev_post(device: Device):
-	print(device)
-	return {"message": "Sucess"}
-	
[e-mail]("/")
-async def root_read():
-	print("Hello")
-	return {"message":"Hello World"}
+from typing import List
+
+from fastapi import Depends, FastAPI, HTTPException
+from sqlalchemy.orm import Session
+from sql_app import crud, models, schemas
+import uvicorn
+from database import SessionLocal, engine
+
+models.Base.metadata.create_all(bind=engine)
+
+app = FastAPI(root_path="/api/v1")
+
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
[e-mail]("/device/", response_model=schemas.Device)
+def create_device(device: schemas.DeviceCreate, db: Session = Depends(get_db)):
+    print(crud.create_device(db=db, device=device))
+
+
[e-mail]("/devices/", response_model=List[schemas.Device])
+def read_devices(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    devices = crud.get_devices(db, skip=skip, limit=limit)
+    return devices
+
+
[e-mail]("/device/{device_id}", response_model=schemas.Device)
+def read_device(device_id: int, db: Session = Depends(get_db)):
+    db_device = crud.get_device(db, device_id=device_id)
+    if db_device is None:
+        raise HTTPException(status_code=404, detail="Device not found")
+    return db_device
+
+
[e-mail]("/usb-logs/", response_model=schemas.USBLog)
+def create_device_logs(log: schemas.USBTempBase, db: Session = Depends(get_db)):
+    dev = crud.find_device(db, log.device)
+    if dev is None:
+        dev = crud.create_device(db=db, device=log.device)
+
+    print(crud.create_device_logs(db=db, item=log, dev_id=dev.id))
+
+
[e-mail]("/logs/", response_model=List[schemas.USBLog])
+def read_logs(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    items = crud.get_logs(db, skip=skip, limit=limit)
+    return items
+
+
[e-mail]("/logs/{device_id}", response_model=List[schemas.USBLog])
+def read_log(device_id: int, db: Session = Depends(get_db)):
+    db_log = crud.get_log(db, device_id=device_id)
+    if db_log is None:
+        raise HTTPException(status_code=404, detail="Logs not found")
+    return db_log
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="192.168.0.22", port=8000)

server/sql_app/models.py

@@ +1 @@
+from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from .database import Base
+
+
+class Device(Base):
+    __tablename__ = "devices"
+
+    id = Column(Integer, primary_key=True, index=True)
+    vendor_id = Column(Integer, index=True, nullable=False)
+    product_id = Column(Integer, index=True, nullable=False)
+    serial_number = Column(String, index=True, nullable=False)
+
+    logs = relationship("USBLog", back_populates="device")
+
+
+class USBLog(Base):
+    __tablename__ = "usb_logs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, index=True, nullable=False)
+    hostname = Column(String, index=True, nullable=False)
+    timestamp = Column(String, index=True, nullable=False)
+    status = Column(String, index=True, nullable=False)
+    device_id = Column(Integer, ForeignKey("devices.id"))
+
+    device = relationship("Device", back_populates="logs")

server/sql_app/schemas.py

@@ +1 @@
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+
+class USBLogBase(BaseModel):
+    timestamp: str
+    status: str
+
+
+class USBLogCreate(USBLogBase):
+    pass
+
+
+class USBLog(USBLogBase):
+    id: int
+    device_id: int
+
+    class Config:
+        orm_mode = True
+
+
+class DeviceBase(BaseModel):
+    vendor_id: int
+    product_id: int
+    serial_number: str
+
+
+class DeviceCreate(DeviceBase):
+    pass
+
+
+class Device(DeviceBase):
+    id: int
+    logs: List[USBLog] = []
+
+    class Config:
+        orm_mode = True
+
+
+class USBTempBase(BaseModel):
+    username: str
+    hostname: str
+    timestamp: str
+    device: DeviceBase
+    status: str
+
+
+class USBTempCreate(USBTempBase):
+    pass
+
+
+class USBTemp(USBTempBase):
+    id: int
+    device_id: int
+
+    class Config:
+        orm_mode = True