Adding bsic tutes for FASTAPI

Author: gauravkoradiya

Body_multiple_parameter.py

@@ -0,0 +1,24 @@
+from typing import Optional
+from fastapi import FastAPI,Body
+from pydantic import BaseModel
+
+app = FastAPI()
+
+class Item(BaseModel):
+    name:str
+    description:Optional[str] = None
+    price : float
+    tax : Optional[float] = None
+
+class User(BaseModel):
+    usename : str
+    fulll_name : Optional[str] = None
+
+
+@app.put('/items/{item_id}')
+async def update_item(item_id:int, item:Item ,user: User, importance:int = Body(...)):
+    results = {"item_id" : item_id,
+               "item" : item,
+               "user" : user,
+               "importance" : importance}
+    return results

Path_Parameter.py

@@ -0,0 +1,11 @@
+from typing import Optional
+from fastapi import FastAPI,Path,Query
+
+app=  FastAPI()
+
+@app.get("/items/{item_id}")
+async def read_items(item_id:int = Path(...,title="title",ge=1, le=100), q: Optional[str] = Query(None, alias="item-query")):
+    results = {"item_id": item_id}
+    if q:
+        results.update({"q": q})
+    return results

Query_Parameter.py

@@ -0,0 +1,37 @@
+from fastapi import FastAPI,Query
+from typing import Optional
+from typing import List
+
+app = FastAPI()
+
+fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
+@app.get("/items/")
+async def read_item(skip : int = 0, limit:int = 10):
+    return fake_items_db[skip:skip+limit]
+
+
+@app.get("/users/{user_id}/items/{item_id}")
+async def read_user_item(user_id: int, item_id: str, q: Optional[str] = None, short: bool = False):
+    item = {"item_id": item_id, "owner_id": user_id}
+    print(item)
+    if q:
+        item.update({"q": q})
+    if short:
+        item.update(
+            {"description": "This is an amazing item that has a long description"}
+        )
+    print(item)
+    return item
+
+@app.get("/item_validation/")
+async def read_items1(q:Optional[str] = Query("fixedquery",max_length=50)):
+    results = {"test" : "a"}
+    if q:
+        results.update({"test1":q})
+    return results
+
+@app.get("/item_list/")
+async def list_parameter(q:List[str]=Query(None)):
+    query_items = {"q": q}
+    print(query_items)
+    return query_items

Request_Body.py

@@ -0,0 +1,26 @@
+from typing import Optional
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+class Item(BaseModel):
+    name:str
+    description: Optional[str] = None
+    price: float
+    tax: Optional[float]
+
+app = FastAPI()
+
+@app.post("/items/")
+async def create_item(item:Item):
+    item_dict = item.dict()
+    if item.tax:
+        price_with_tax = item.price + item.tax
+        item_dict.update({"price_with_tax" : price_with_tax})
+    return item_dict
+
+@app.put("/items/{item_id}")
+async def create_item1(item_id: int, item: Item, q: Optional[str] = None):
+    result = {"item_id": item_id, **item.dict()}
+    if q:
+        result.update({"q": q})
+    return result

async_practice/FastAPI_async_await.py

@@ -0,0 +1,28 @@
+import asyncio
+import time
+
+async def sleep():
+    print(f'Time: {time.time() - start:.2f}')
+    time.sleep(1)
+
+
+async def sum(name, numbers):
+    total = 0
+    for number in numbers:
+        print(f'Task {name}: Computing {total}+{number}')
+        await sleep()
+        total += number
+    print(f'Task {name}: Sum = {total}\n')
+
+start = time.time()
+loop=  asyncio.get_event_loop()
+tasks = [loop.create_task(sum("A", [1, 2])),
+         loop.create_task(sum("B", [1, 2, 3, 4]))]
+loop.run_until_complete(asyncio.wait(tasks))
+loop.close()
+
+end = time.time()
+print(f'Time: {end-start:.2f} sec')
+
+
+

async_practice/async_await_execution_wrong.py

@@ -0,0 +1,32 @@
+import asyncio
+import time
+"""
+sleep function in normal and wrong are the "same". They "sleep" without allowing others to use the resources. Whereas this allows access to the resources when it is asleep.
+In case 2(wrong) we added async to the normal function. However the event loop will run it without interruption. Why? Because we didn't tell where the loop is allowed to interrupt your function to run another task.
+
+In case 3 we told the event loop exactly where to interrupt the function to run another task. Where exactly?
+"""
+async def sleep():
+    print(f'Time: {time.time() - start:.2f}')
+    await asyncio.sleep(1)
+
+async def sum(name, numbers):
+    total = 0
+    for number in numbers:
+        print(f'Task {name}: Computing {total}+{number}')
+        await sleep()
+        total += number
+    print(f'Task {name}: Sum = {total}\n')
+
+start = time.time()
+
+loop = asyncio.get_event_loop()
+tasks = [
+    loop.create_task(sum("A", [1, 2])),
+    loop.create_task(sum("B", [1, 2, 3])),
+]
+loop.run_until_complete(asyncio.wait(tasks))
+loop.close()
+
+end = time.time()
+print(f'Time: {end-start:.2f} sec')

async_practice/async_await_right_execution.py

@@ -0,0 +1,17 @@
+import asyncio
+
+async def factorial(n):
+    f = 1
+    for i in range(2, n + 1):
+        print(f"Computing factorial({n}), currently i={i}...")
+        await asyncio.sleep(1)
+        f *= i
+    return f
+
+async def main():
+    L = await asyncio.gather(factorial(2), factorial(3), factorial(4))
+    print(L)  # [2, 6, 24]
+
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(main())

async_practice/factorial_async.py

@@ -0,0 +1,28 @@
+import time
+
+
+"""
+This is normal sysn excecution.
+Examples from : https://stackoverflow.com/questions/50757497/simplest-async-await-example-possible-in-python
+
+"""
+
+def sleep():
+    print(f'Time: {time.time() - start:.2f}')
+    time.sleep(1)
+
+def sum(name, numbers):
+    total = 0
+    for number in numbers:
+        print(f'Task {name}: Computing {total}+{number}')
+        sleep()
+        total += number
+    print(f'Task {name}: Sum = {total}\n')
+
+start = time.time()
+task = [
+    sum("A", [1, 2]),
+    sum("B", [1, 2, 3]),
+]
+end = time.time()
+print(f'Time: {end-start:.2f} sec')

async_practice/normal_sync_execution.py

@@ -0,0 +1,41 @@
+"""
+https://stackoverflow.com/a/50758540/7786625
+
+why do you need asyncio, not just plain code?
+Answer is - asyncio allows you to get performance benefit when you parallelize I/O blocking operations (like reading/writing to network).
+
+
+example that uses asyncio.sleep to imitate I/O blocking operation and asyncio.gather that shows how you can run multiple blocking operations concurrently:
+"""
+
+import asyncio
+import time
+
+
+async def io_related(name):
+    print(f'{name} started')
+    await asyncio.sleep(1)
+    # time.sleep(1)
+    print(f'{name} finished')
+
+
+async def main():
+    await asyncio.gather(
+        io_related('first'),
+        io_related('second'),
+        io_related('Third'),
+        io_related('Four'),
+        io_related('Five'),
+        io_related('Six'),
+        io_related('Seven'),
+        io_related('Eight'),
+        io_related('Nine'),
+
+    )
+
+
+if __name__ ==  '__main__':
+    loop = asyncio.get_event_loop()
+    start = time.time()
+    loop.run_until_complete(main())
+    print(f'{time.time()-start} second.')