Revize 6c1e92e3

re #9425 Struktury upraveny dle vize. Endpointy rozděleny do souboru podle funkcionality.

server/sql_app/api/devices.py

@@ +1 @@
+from typing import List
+
+from fastapi import Depends, FastAPI, HTTPException, APIRouter
+from sqlalchemy.orm import Session
+from ...sql_app import crud, models, schemas
+from ..database import SessionLocal, engine
+
+models.Base.metadata.create_all(bind=engine)
+
+device = APIRouter()
+
+# 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

server/sql_app/api/licenses.py

@@ +1 @@
+from typing import List
+
+from fastapi import Depends, FastAPI, HTTPException, APIRouter
+from sqlalchemy.orm import Session
+from ...sql_app import crud, models, schemas
+from ..database import SessionLocal, engine
+
+models.Base.metadata.create_all(bind=engine)
+
+licenses = APIRouter()
+
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
[e-mail]("/license/", response_model=schemas.License)
+def create_license(license: schemas.LicenseCreate, db: Session = Depends(get_db)):
+    print(crud.create_license(db=db, license=license))
+
+
[e-mail]("/licenses/", response_model=List[schemas.License])
+def read_licenses(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    licenses = crud.get_licenses(db, skip=skip, limit=limit)
+    return licenses
+
+
[e-mail]("/device_license/", response_model=schemas.DeviceLicense)
+def create_device_license(device_license: schemas.DeviceLicenseCreate, db: Session = Depends(get_db)):
+    print(crud.create_device_license(db=db, device_license=device_license))

server/sql_app/api/pcs.py

@@ +1 @@
+from typing import List
+from fastapi import Depends, FastAPI, HTTPException, APIRouter
+from sqlalchemy.orm import Session
+from ...sql_app import crud, models, schemas
+from ..database import SessionLocal, engine
+
+models.Base.metadata.create_all(bind=engine)
+
+pcs = APIRouter()
+
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
[e-mail]("/pc/", response_model=schemas.PC)
+def create_pc(pc: schemas.PCCreate, db: Session = Depends(get_db)):
+    print(crud.create_pc(db=db, user=pc.username, host=pc.hostname))
+
+
[e-mail]("/pcs/", response_model=List[schemas.PC])
+def read_pcs(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    pcs = crud.get_pcs(db, skip=skip, limit=limit)
+    return pcs
+
+
[e-mail]("/pc/{pc_id}", response_model=schemas.PC)
+def read_pc(pc_id: int, db: Session = Depends(get_db)):
+    db_pc = crud.get_pc(db, pc_id=pc_id)
+    if db_pc is None:
+        raise HTTPException(status_code=404, detail="Device not found")
+    return db_pc

server/sql_app/api/usb_logs.py

@@ +1 @@
+from typing import List
+from fastapi import Depends, FastAPI, HTTPException, APIRouter, Form
+from sqlalchemy.orm import Session
+from ...sql_app import crud, models, schemas
+from ..database import SessionLocal, engine
+
+models.Base.metadata.create_all(bind=engine)
+
+usblogs = APIRouter()
+
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
[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)
+    pc = crud.find_pc(db, log.username, log.hostname)
+    if pc is None:
+        pc = crud.create_pc(db=db, user=log.username, host=log.hostname)
+
+    print(crud.create_device_logs(db=db, item=log, dev_id=dev.id, pc_id=pc.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/crud.py

@@ -26 +26 @@
     return db_device
 
 
+def get_license(db: Session, license_id: int):
+    return db.query(models.License).filter(models.License.id == license_id).first()
+
+
+def get_licenses(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.License).offset(skip).limit(limit).all()
+
+
+def create_license(db: Session, license: schemas.LicenseBase):
+    db_license = models.License(name=license.name, expiration_date=license.expiration_date)
+    db.add(db_license)
+    db.commit()
+    db.refresh(db_license)
+    return db_license
+
+
+def create_device_license(db: Session, device_license: schemas.DeviceLicenseBase):
+    db_device_license = models.DeviceLicense(device_id=device_license.device_id, license_id=device_license.license_id,
+                                             assigned_datetime=device_license.assigned_datetime)
+    db.add(db_device_license)
+    db.commit()
+    db.refresh(db_device_license)
+    return db_device_license
+
+
+def get_pc(db: Session, pc_id: int):
+    return db.query(models.PC).filter(models.PC.id == pc_id).first()
+
+
+def get_pcs(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.PC).offset(skip).limit(limit).all()
+
+
+def find_pc(db: Session, username: str, hostname: str):
+    return db.query(models.PC).filter(models.PC.username == username and
+                                      models.PC.hostname == hostname).first()
+
+
+def create_pc(db: Session, user: str, host: str):
+    db_pc = models.PC(username=user, hostname=host)
+    db.add(db_pc)
+    db.commit()
+    db.refresh(db_pc)
+    return db_pc
+
+
 def get_logs(db: Session, skip: int = 0, limit: int = 100):
     return db.query(models.USBLog).offset(skip).limit(limit).all()
 
@@ -34 +80 @@
     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)
+def create_device_logs(db: Session, item: schemas.USBTempBase, dev_id: int, pc_id: int):
+    db_log = models.USBLog(pc_id=pc_id, timestamp=item.timestamp, status=item.status, device_id=dev_id)
     db.add(db_log)
     db.commit()
     db.refresh(db_log)

server/sql_app/main.py

@@ -1 @@
-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
+from server.sql_app.api.devices import device
+from server.sql_app.api.licenses import licenses
+from server.sql_app.api.pcs import pcs
+from server.sql_app.api.usb_logs import usblogs
+from fastapi import FastAPI
 
-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
-
+app.include_router(device)
+app.include_router(licenses)
+app.include_router(pcs)
+app.include_router(usblogs)
 
 if __name__ == "__main__":
     uvicorn.run(app, host="192.168.0.22", port=8000)
+

server/sql_app/models.py

@@ -13 +13 @@
     serial_number = Column(String, index=True, nullable=False)
 
     logs = relationship("USBLog", back_populates="device")
+    licenses = relationship("DeviceLicense", back_populates="device_lic")
 
 
 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)
+    pc_id = Column(Integer, ForeignKey("pc.id"))
     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")
+    pc = relationship("PC", back_populates="logs_pc")
+
+
+class License(Base):
+    __tablename__ = "licenses"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, index=True, nullable=False)
+    expiration_date = Column(String, index=True, nullable=False)
+
+    devices = relationship("DeviceLicense", back_populates="licenses")
+
+
+class DeviceLicense(Base):
+    __tablename__ = "devices_licenses"
+
+    id = Column(Integer, primary_key=True, index=True)
+    device_id = Column(Integer, ForeignKey("devices.id"))
+    license_id = Column(Integer, ForeignKey("licenses.id"))
+    assigned_datetime = Column(String, index=True, nullable=False)
+
+    device_lic = relationship("Device", back_populates="licenses")
+    licenses = relationship("License", back_populates="devices")
+
+
+class PC(Base):
+    __tablename__ = "pc"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, index=True, nullable=False)
+    hostname = Column(String, index=True, nullable=False)
+    logs_pc = relationship("USBLog", back_populates="pc")

server/sql_app/schemas.py

@@ -4 +4 @@
 
 
 
+class DeviceLicenseBase(BaseModel):
+    device_id: int
+    license_id: int
+    assigned_datetime: str
+
+
+class DeviceLicenseCreate(DeviceLicenseBase):
+    pass
+
+
+class DeviceLicense(DeviceLicenseCreate):
+    id: int
+
+    class Config:
+        orm_mode = True
+
+
 class USBLogBase(BaseModel):
     timestamp: str
     status: str
@@ -16 +33 @@
 class USBLog(USBLogBase):
     id: int
     device_id: int
+    pc_id: int
 
     class Config:
         orm_mode = True
@@ -34 +52 @@
 class Device(DeviceBase):
     id: int
     logs: List[USBLog] = []
+    licenses: List[DeviceLicense] = []
+
+    class Config:
+        orm_mode = True
+
+
+class PCBase(BaseModel):
+    username: str
+    hostname: str
+
+
+class PCCreate(PCBase):
+    pass
+
+
+class PC(PCCreate):
+    id: int
+    logs_pc: List[USBLog] = []
+
+    class Config:
+        orm_mode = True
+
+
+class LicenseBase(BaseModel):
+    name: str
+    expiration_date: str
+
+
+class LicenseCreate(LicenseBase):
+    pass
+
+
+class License(LicenseCreate):
+    id: int
+    devices: List[DeviceLicense] = []
 
     class Config:
         orm_mode = True