generated from render-examples/fastapi
-
Notifications
You must be signed in to change notification settings - Fork 1
/
extra.py
105 lines (91 loc) · 2.79 KB
/
extra.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
import json
# Define a function to read and load JSON data from a file
def load_json(file_path):
with open(file_path, 'r') as file:
data = json.load(file)
return data
# Example usage
file_path = 'test.json' # Replace 'example.json' with the path to your JSON file
json_data = load_json(file_path)
# Access the loaded JSON data
print(json_data["initial_songs"])
##################################
{
"company": "TechCorp",
"location": "New York",
"employees": [
{
"id": 1,
"name": "John Doe",
"position": "Software Engineer",
"department": "Engineering",
"salary": 80000
},
{
"id": 2,
"name": "Jane Smith",
"position": "Data Scientist",
"department": "Data Science",
"salary": 90000
},
{
"id": 3,
"name": "Alice Johnson",
"position": "Product Manager",
"department": "Product Management",
"salary": 100000
}
]
}
#################################
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict
import json
app = FastAPI()
class Employee(BaseModel):
id: int
name: str
position: str
department: str
salary: float
# Load initial data from JSON file
with open("employees.json", "r") as file:
employees_data = json.load(file)
# Helper function to save data back to JSON file
def save_data(data: Dict):
with open("employees.json", "w") as file:
json.dump(data, file, indent=2)
@app.get("/employees/", response_model=List[Employee])
async def read_employees():
return employees_data.get("employees", [])
@app.post("/employees/")
async def create_employee(employee: Employee):
employees = employees_data.get("employees", [])
employees.append(employee.dict())
employees_data["employees"] = employees
save_data(employees_data)
return {"message": "Employee created successfully"}
@app.put("/employees/{employee_id}")
async def update_employee(employee_id: int, employee: Employee):
employees = employees_data.get("employees", [])
for emp in employees:
if emp["id"] == employee_id:
emp.update(employee.dict())
break
else:
raise HTTPException(status_code=404, detail="Employee not found")
save_data(employees_data)
return {"message": "Employee updated successfully"}
@app.delete("/employees/{employee_id}")
async def delete_employee(employee_id: int):
employees = employees_data.get("employees", [])
for i, emp in enumerate(employees):
if emp["id"] == employee_id:
del employees[i]
break
else:
raise HTTPException(status_code=404, detail="Employee not found")
employees_data["employees"] = employees
save_data(employees_data)
return {"message": "Employee deleted successfully"}