Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4import datetime
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10plots = {}
11harvests = {}
12soil_data = {}
13next_user_id = 1
14next_plot_id = 1
15next_harvest_id = 1
16next_soil_id = 1
17 
18class SignupRequest(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class PlotCreate(BaseModel):
27    name: str
28    location: str
29 
30class HarvestCreate(BaseModel):
31    plot_id: int
32    crop: str
33    weight_kg: float
34 
35class SoilDataCreate(BaseModel):
36    plot_id: int
37    ph: float
38    moisture_percent: float
39    nitrogen_ppm: float
40 
41def get_user_from_token(authorization: str = Header(None)):
42    if not authorization:
43        raise HTTPException(status_code=401, detail="Authorization header required")
44    token = authorization.replace("Bearer ", "")
45    user = tokens.get(token)
46    if not user:
47        raise HTTPException(status_code=401, detail="Invalid token")
48    return user
49 
50@app.post("/signup")
51def signup(req: SignupRequest):
52    global next_user_id
53    for u in users.values():
54        if u["username"] == req.username:
55            raise HTTPException(status_code=400, detail="Username taken")
56    uid = next_user_id
57    next_user_id += 1
58    users[uid] = {"id": uid, "username": req.username, "password": req.password}
59    return {"id": uid, "username": req.username}
60 
61@app.post("/login")
62def login(req: LoginRequest):
63    for u in users.values():
64        if u["username"] == req.username and u["password"] == req.password:
65            token = secrets.token_hex(16)
66            tokens[token] = u
67            return {"token": token}
68    raise HTTPException(status_code=401, detail="Invalid credentials")
69 
70@app.get("/plots/{plot_id}")
71def get_plot(plot_id: int, authorization: str = Header(None)):
72    get_user_from_token(authorization)
73    plot = plots.get(plot_id)
74    if not plot:
75        raise HTTPException(status_code=404, detail="Plot not found")
76    return plot
77 
78@app.post("/plots")
79def create_plot(plot: PlotCreate, authorization: str = Header(None)):
80    user = get_user_from_token(authorization)
81    global next_plot_id
82    pid = next_plot_id
83    next_plot_id += 1
84    plots[pid] = {"id": pid, "name": plot.name, "location": plot.location, "claimed_by": user["id"]}
85    return plots[pid]
86 
87@app.get("/harvests/{harvest_id}")
88def get_harvest(harvest_id: int, authorization: str = Header(None)):
89    get_user_from_token(authorization)
90    harvest = harvests.get(harvest_id)
91    if not harvest:
92        raise HTTPException(status_code=404, detail="Harvest not found")
93    return harvest
94 
95@app.post("/harvests")
96def create_harvest(harvest: HarvestCreate, authorization: str = Header(None)):
97    user = get_user_from_token(authorization)
98    if harvest.plot_id not in plots:
99        raise HTTPException(status_code=404, detail="Plot not found")
100    if plots[harvest.plot_id]["claimed_by"] != user["id"]:
101        raise HTTPException(status_code=403, detail="You do not own this plot")
102    global next_harvest_id
103    hid = next_harvest_id
104    next_harvest_id += 1
105    harvests[hid] = {
106        "id": hid,
107        "plot_id": harvest.plot_id,
108        "crop": harvest.crop,
109        "weight_kg": harvest.weight_kg,
110        "date": datetime.datetime.utcnow().isoformat()
111    }
112    return harvests[hid]
113 
114@app.get("/soil/{soil_id}")
115def get_soil(soil_id: int, authorization: str = Header(None)):
116    get_user_from_token(authorization)
117    entry = soil_data.get(soil_id)
118    if not entry:
119        raise HTTPException(status_code=404, detail="Soil record not found")
120    return entry
121 
122@app.post("/soil")
123def create_soil(soil: SoilDataCreate, authorization: str = Header(None)):
124    user = get_user_from_token(authorization)
125    if soil.plot_id not in plots:
126        raise HTTPException(status_code=404, detail="Plot not found")
127    if plots[soil.plot_id]["claimed_by"] != user["id"]:
128        raise HTTPException(status_code=403, detail="You do not own this plot")
129    global next_soil_id
130    sid = next_soil_id
131    next_soil_id += 1
132    soil_data[sid] = {
133        "id": sid,
134        "plot_id": soil.plot_id,
135        "ph": soil.ph,
136        "moisture_percent": soil.moisture_percent,
137        "nitrogen_ppm": soil.nitrogen_ppm,
138        "recorded_at": datetime.datetime.utcnow().isoformat()
139    }
140    return soil_data[sid]

1fastapi
2uvicorn