fastapi_basics/app.py at main · natrajexplore/fastapi_basics · GitHub

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
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="Task cd nager API")

# ----------------------------
# Model
# ----------------------------

class Task(BaseModel):
    title: str
    description: Optional[str] = None
    completed: bool = False


# Temporary database
tasks = {}


# ----------------------------
# Home
# ----------------------------

@app.get("/")
def home():
    return {"message": "Welcome to FastAPI Task Manager"}


# ----------------------------
# Create Task
# ----------------------------

@app.post("/tasks/{task_id}")
def create_task(task_id: int, task: Task):

    if task_id in tasks:
        raise HTTPException(
            status_code=400,
            detail="Task already exists"
        )

    tasks[task_id] = task

    return {
        "message": "Task Created Successfully",
        "task": task
    }


# ----------------------------
# Read All Tasks
# ----------------------------

@app.get("/tasks")
def get_tasks():
    return tasks


# ----------------------------
# Read One Task
# ----------------------------

@app.get("/tasks/{task_id}")
def get_task(task_id: int):

    if task_id not in tasks:
        raise HTTPException(
            status_code=404,
            detail="Task Not Found"
        )

    return tasks[task_id]


# ----------------------------
# Update Task
# ----------------------------

@app.put("/tasks/{task_id}")
def update_task(task_id: int, task: Task):

    if task_id not in tasks:
        raise HTTPException(
            status_code=404,
            detail="Task Not Found"
        )

    tasks[task_id] = task

    return {
        "message": "Task Updated",
        "task": task
    }


# ----------------------------
# Delete Task
# ----------------------------

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):

    if task_id not in tasks:
        raise HTTPException(
            status_code=404,
            detail="Task Not Found"
        )

    del tasks[task_id]

    return {
        "message": "Task Deleted"
    }


# ----------------------------
# Search Tasks
# ----------------------------

@app.get("/search")
def search(completed: bool = False):

    result = {}

    for key, value in tasks.items():
        if value.completed == completed:
            result[key] = value

    return result