Browse the data

1from fastapi import FastAPI, HTTPException, Header, UploadFile, File, Form
2from pydantic import BaseModel
3from typing import Optional, List
4import uuid
5import shutil
6import os
7from datetime import datetime
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13tickets = {}
14ticket_id_counter = 1
15uploads_dir = "uploads"
16os.makedirs(uploads_dir, exist_ok=True)
17 
18class SignupRequest(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class TicketCreate(BaseModel):
27    description: str
28    priority: Optional[str] = "medium"
29    assigned_worker: Optional[str] = None
30    status: Optional[str] = "open"
31 
32class TicketUpdate(BaseModel):
33    description: Optional[str] = None
34    priority: Optional[str] = None
35    assigned_worker: Optional[str] = None
36    status: Optional[str] = None
37 
38def get_user_from_token(authorization: str = Header(None)):
39    if not authorization:
40        raise HTTPException(status_code=401, detail="No auth header")
41    token = authorization.replace("Bearer ", "")
42    if token not in tokens:
43        raise HTTPException(status_code=401, detail="Invalid token")
44    return tokens[token]
45 
46@app.post("/signup")
47def signup(req: SignupRequest):
48    if req.username in users:
49        raise HTTPException(status_code=400, detail="User exists")
50    users[req.username] = {"username": req.username, "password": req.password}
51    return {"message": "User created"}
52 
53@app.post("/login")
54def login(req: LoginRequest):
55    if req.username not in users or users[req.username]["password"] != req.password:
56        raise HTTPException(status_code=401, detail="Invalid credentials")
57    token = str(uuid.uuid4())
58    tokens[token] = req.username
59    return {"token": token}
60 
61@app.get("/tickets/{ticket_id}")
62def get_ticket(ticket_id: int, authorization: str = Header(None)):
63    get_user_from_token(authorization)
64    if ticket_id not in tickets:
65        raise HTTPException(status_code=404, detail="Ticket not found")
66    return tickets[ticket_id]
67 
68@app.post("/tickets")
69def create_ticket(req: TicketCreate, authorization: str = Header(None)):
70    get_user_from_token(authorization)
71    global ticket_id_counter
72    ticket = {
73        "id": ticket_id_counter,
74        "description": req.description,
75        "priority": req.priority,
76        "assigned_worker": req.assigned_worker,
77        "status": req.status,
78        "images": [],
79        "history": [{"action": "created", "timestamp": datetime.now().isoformat()}]
80    }
81    tickets[ticket_id_counter] = ticket
82    ticket_id_counter += 1
83    return ticket
84 
85@app.patch("/tickets/{ticket_id}")
86async def update_ticket(
87    ticket_id: int,
88    description: Optional[str] = Form(None),
89    priority: Optional[str] = Form(None),
90    assigned_worker: Optional[str] = Form(None),
91    status: Optional[str] = Form(None),
92    images: List[UploadFile] = File(None),
93    authorization: str = Header(None)
94):
95    get_user_from_token(authorization)
96    if ticket_id not in tickets:
97        raise HTTPException(status_code=404, detail="Ticket not found")
98    ticket = tickets[ticket_id]
99    changes = []
100    if description is not None and description != ticket["description"]:
101        changes.append(f"description: '{ticket['description']}' -> '{description}'")
102        ticket["description"] = description
103    if priority is not None and priority != ticket["priority"]:
104        changes.append(f"priority: '{ticket['priority']}' -> '{priority}'")
105        ticket["priority"] = priority
106    if assigned_worker is not None and assigned_worker != ticket["assigned_worker"]:
107        changes.append(f"assigned_worker: '{ticket['assigned_worker']}' -> '{assigned_worker}'")
108        ticket["assigned_worker"] = assigned_worker
109    if status is not None and status != ticket["status"]:
110        changes.append(f"status: '{ticket['status']}' -> '{status}'")
111        ticket["status"] = status
112    if images:
113        for img in images:
114            file_path = os.path.join(uploads_dir, f"{ticket_id}_{img.filename}")
115            with open(file_path, "wb") as buffer:
116                shutil.copyfileobj(img.file, buffer)
117            ticket["images"].append(file_path)
118            changes.append(f"image added: {img.filename}")
119    if changes:
120        ticket["history"].append({
121            "action": "updated",
122            "changes": changes,
123            "timestamp": datetime.now().isoformat()
124        })
125    return ticket

1fastapi
2uvicorn
3python-multipart