-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapi.py
More file actions
112 lines (97 loc) · 3.11 KB
/
Copy pathapi.py
File metadata and controls
112 lines (97 loc) · 3.11 KB
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
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import uvicorn
# Initialize FastAPI app
app = FastAPI(
title="My API Service",
description="A sample REST API service with Swagger documentation",
version="1.0.0"
)
# Data models
class Item(BaseModel):
id: Optional[int] = None
name: str
description: Optional[str] = None
price: float
in_stock: bool = True
class User(BaseModel):
id: Optional[int] = None
username: str
email: str
full_name: Optional[str] = None
# In-memory storage
items_db = []
users_db = []
item_id_counter = 1
user_id_counter = 1
# Root endpoint
@app.get("/")
async def root():
"""Welcome endpoint"""
return {"message": "Welcome to the API. Visit /docs for Swagger UI"}
# Item endpoints
@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
"""Create a new item"""
global item_id_counter
item.id = item_id_counter
item_id_counter += 1
items_db.append(item)
return item
@app.get("/items/", response_model=List[Item], tags=["items"])
async def get_items():
"""Get all items"""
return items_db
@app.get("/items/{item_id}", response_model=Item, tags=["items"])
async def get_item(item_id: int):
"""Get a specific item by ID"""
for item in items_db:
if item.id == item_id:
return item
raise HTTPException(status_code=404, detail="Item not found")
@app.put("/items/{item_id}", response_model=Item, tags=["items"])
async def update_item(item_id: int, item: Item):
"""Update an existing item"""
for idx, existing_item in enumerate(items_db):
if existing_item.id == item_id:
item.id = item_id
items_db[idx] = item
return item
raise HTTPException(status_code=404, detail="Item not found")
@app.delete("/items/{item_id}", tags=["items"])
async def delete_item(item_id: int):
"""Delete an item"""
for idx, item in enumerate(items_db):
if item.id == item_id:
items_db.pop(idx)
return {"message": "Item deleted successfully"}
raise HTTPException(status_code=404, detail="Item not found")
# User endpoints
@app.post("/users/", response_model=User, tags=["users"])
async def create_user(user: User):
"""Create a new user"""
global user_id_counter
user.id = user_id_counter
user_id_counter += 1
users_db.append(user)
return user
@app.get("/users/", response_model=List[User], tags=["users"])
async def get_users():
"""Get all users"""
return users_db
@app.get("/users/{user_id}", response_model=User, tags=["users"])
async def get_user(user_id: int):
"""Get a specific user by ID"""
for user in users_db:
if user.id == user_id:
return user
raise HTTPException(status_code=404, detail="User not found")
# Health check endpoint
@app.get("/health", tags=["system"])
async def health_check():
"""Check if the service is running"""
return {"status": "healthy", "service": "running"}
if __name__ == "__main__":
# Run the service
uvicorn.run(app, host="0.0.0.0", port=8000)