Revize 54840605

First api implementation. Lacks postgresql connection and web templates

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 create_device(db: Session, device: schemas.DeviceCreate):
+    db_device = models.Device(vendor_id=device.vendor_id, product_id=device.product_id)
+    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.USBLogCreate, dev_id: int):
+    db_log = models.USBLog(**item.dict(), 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
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+

server/sql_app/main.py

@@ +1 @@
+from typing import List
+
+from fastapi import Depends, FastAPI, HTTPException
+from sqlalchemy.orm import Session
+
+from . import crud, models, schemas
+from .database import SessionLocal, engine
+
+models.Base.metadata.create_all(bind=engine)
+
+
+app = FastAPI()
+    
+# 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)):
+    return 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]("/api/v1/logs/{device_id}/", response_model=schemas.USBLog)
+def create_device_logs(device_id: int, log: schemas.USBLogCreate, db: Session = Depends(get_db)):
+    return crud.create_device_logs(db=db, item=log, dev_id=device_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
+

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__ = "usb_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)
+
+    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("usb_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):
+    username: str
+    hostname: str
+    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
+
+
+class DeviceCreate(DeviceBase):
+    pass
+
+
+class Device(DeviceBase):
+    id: int
+    logs: List[USBLog] = []
+    class Config:
+        orm_mode = True
+