forked from farmOS/farmOS-aggregator
-
Notifications
You must be signed in to change notification settings - Fork 0
/
logs.py
134 lines (102 loc) · 2.93 KB
/
logs.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
from typing import List
from fastapi import APIRouter, Body, Depends, HTTPException, Query
from starlette.requests import Request
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.api.utils.db import get_db
from app.api.utils.farms import get_active_farms_url_or_list, get_farm_client, ClientError
from app.schemas.farm import Farm
router = APIRouter()
# /farms/logs/ endpoint for accessing farmOS logs
class Log(BaseModel):
class Config:
extra = 'allow'
name: str
type: str
class LogUpdate(BaseModel):
class Config:
extra = 'allow'
id: int
@router.get("/")
def get_all_farm_logs(
request: Request,
farm_list: List[Farm] = Depends(get_active_farms_url_or_list),
db: Session = Depends(get_db),
):
query_params = {**request.query_params}
query_params.pop('farm_id', None)
query_params.pop('farm_url', None)
data = {}
for farm in farm_list:
data[farm.id] = []
# Get a farmOS client.
try:
farm_client = get_farm_client(db_session=db, farm=farm)
except ClientError:
continue
# Make the request.
try:
data[farm.id] = data[farm.id] + farm_client.log.get(filters=query_params)['list']
except:
continue
return data
@router.post("/")
def create_farm_logs(
log: Log,
farm_list: List[Farm] = Depends(get_active_farms_url_or_list),
db: Session = Depends(get_db),
):
data = {}
for farm in farm_list:
data[farm.id] = []
# Get a farmOS client.
try:
farm_client = get_farm_client(db_session=db, farm=farm)
except ClientError:
continue
# Make the request.
try:
data[farm.id].append(farm_client.log.send(payload=log.dict()))
except:
continue
return data
@router.put("/")
def update_farm_logs(
log: LogUpdate,
farm_list: List[Farm] = Depends(get_active_farms_url_or_list),
db: Session = Depends(get_db),
):
data = {}
for farm in farm_list:
data[farm.id] = []
# Get a farmOS client.
try:
farm_client = get_farm_client(db_session=db, farm=farm)
except ClientError:
continue
# Make the request.
try:
data[farm.id].append(farm_client.log.send(payload=log.dict()))
except:
continue
return data
@router.delete("/")
def delete_farm_logs(
id: int,
farm_list: List[Farm] = Depends(get_active_farms_url_or_list),
db: Session = Depends(get_db),
):
data = {}
for farm in farm_list:
data[farm.id] = []
# Get a farmOS client.
try:
farm_client = get_farm_client(db_session=db, farm=farm)
except ClientError:
continue
# Make the request.
try:
data[farm.id].append(farm_client.log.delete(id=id))
except:
continue
return data