Projekt

Obecné

Profil

« Předchozí | Další » 

Revize d446b62c

Přidáno uživatelem Matěj Zeman před více než 2 roky(ů)

re #9304 HelloWorld API for sending data defined in USBLog class. Returns simple sucess message

Zobrazit rozdíly:

server/sql_app/main.py
1
from typing import List
2

  
3
from fastapi import Depends, FastAPI, HTTPException
4
from sqlalchemy.orm import Session
5

  
6
from . import crud, models, schemas
7
from .database import SessionLocal, engine
8

  
9
models.Base.metadata.create_all(bind=engine)
10

  
1
from fastapi import FastAPI
2
import fastapi
3
from pydantic import BaseModel
4
from typing import Optional, List
11 5

  
12 6
app = FastAPI()
13
    
14
# Dependency
15
def get_db():
16
    db = SessionLocal()
17
    try:
18
        yield db
19
    finally:
20
        db.close()
21

  
22

  
23
@app.post("/device/", response_model=schemas.Device)
24
def create_device(device: schemas.DeviceCreate, db: Session = Depends(get_db)):
25
    return crud.create_device(db=db, device=device)
26

  
27
@app.get("/devices/", response_model=List[schemas.Device])
28
def read_devices(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
29
    devices = crud.get_devices(db, skip=skip, limit=limit)
30
    return devices
31

  
32

  
33
@app.get("/device/{device_id}", response_model=schemas.Device)
34
def read_device(device_id: int, db: Session = Depends(get_db)):
35
    db_device = crud.get_device(db, device_id=device_id)
36
    if db_device is None:
37
        raise HTTPException(status_code=404, detail="Device not found")
38
    return db_device
39

  
40

  
41
@app.post("/api/v1/logs/{device_id}/", response_model=schemas.USBLog)
42
def create_device_logs(device_id: int, log: schemas.USBLogCreate, db: Session = Depends(get_db)):
43
    return crud.create_device_logs(db=db, item=log, dev_id=device_id)
44

  
45

  
46
@app.get("/logs/", response_model=List[schemas.USBLog])
47
def read_logs(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
48
    items = crud.get_logs(db, skip=skip, limit=limit)
49
    return items
50

  
51
@app.get("/logs/{device_id}", response_model=List[schemas.USBLog])
52
def read_log(device_id: int, db: Session = Depends(get_db)):
53
    db_log = crud.get_log(db, device_id=device_id)
54
    if db_log is None:
55
        raise HTTPException(status_code=404, detail="Logs not found")
56
    return db_log
57 7

  
8
class Device(BaseModel):
9
	vendor_id: int
10
	product_id: int
11
	serial_number: int
12

  
13
class USBLog(BaseModel):
14
	username: str
15
	hostname: str
16
	timestamp: str
17
	device: Device
18
	status:str
19

  
20
@app.post("/api/v1/usb-logs/")
21
async def root_post(usb_log: USBLog):
22
	print(usb_log)
23
	return {"message": "Sucess"}
24
	
25
@app.get("/")
26
async def root_read():
27
	print("Hello")
28
	return {"message":"Hello World"}

Také k dispozici: Unified diff