내맘대로 Fastapi Document summary[5]

 

# You only give Depends a single parameter.
# This parameter must be something like a function.
# async 안에 def or def 안에 async 다 가능

from typing import Optional
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

 

 

# 위와 같이 return으로 dict를 해도 되고, class를 선언해도 된다.

class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

# commons: CommonQueryParams = Depends(CommonQueryParams)에서 class가 중복되니
# -> commons=Depends(CommonQueryParams) 가능.
# 하지만 코드 완벽성을 위해 type을 적어주는 것이 낫다는 docs의 말.
# 그래서 commons: CommonQueryParams = Depends() 이렇게 쓰는게 좋다.

 

 

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: Optional[str] = None):
    return q

def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}

# deep dependency
# 위 코드는 str이 넘어오면 그걸 cookie로 아니면 default cookie값을 쓰겠다.

 

 

# If one of your dependencies is declared multiple times for the same path operation, 
# for example, multiple dependencies have a common sub-dependency
# And it will save the returned value in a "cache" and pass it to all the "dependants" 
# that need it in that specific request, instead of calling the dependency multiple times 
# for the same request.

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
    return {"fresh_value": fresh_value}
    
# 그러니까 dependency 말처럼 요청 들어올 때 dependant instance를 독립적이고, shared 하지 않도록
# 하기 위해서는 use_cache=False로 넘겨주라는 뜻.

 

 

from fastapi import Depends, FastAPI, HTTPException

async def verify_age(commons: CommonQueryParams = Depends(CommonQueryParams)):
    if commons.age < 18:
        raise HTTPException(status_code=400, detail="Requires adult supervision")


async def verify_admin(commons: CommonQueryParams = Depends(CommonQueryParams)):
    if commons.id != 'ID0001':
        raise HTTPException(status_code=400, detail="Requires admin access")


@app.get("/user/", dependencies=[Depends(verify_age)])
async def user():
    return {"message": "Successfully accessed user page"}


@app.get("/admin/", dependencies=[Depends(verify_admin), Depends(verify_age)])
async def admin():
    return {"message": "Successfully accessed admin page"}
    
# http://localhost:8000/admin/?id=ID0001&name=wfng&age=15
# -> {"detail":"Requires adult supervision"}
# http://localhost:8000/admin/?id=ID0002&name=wfng&age=15
# -> {"detail":"Requires admin access"}
# decorator에서의 depends는 return값을 사용 안한다.

 

 

app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

# global dependency

 

 

# yield 예시
async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()
# yield는 yield 시점에서 해당값을 반환하고 멈춰있다가 다시 받으면 그 다음 코드 실행.
# 즉 yield는 generator 역할.

def number_generator(n):
    print("Function Start")
    while n <6:
        yield n
        n+= 1
    print("Function End")
     
if __name__== "__main__":
    for i in number_generator(0):
        print(i)

 

 

# 다음과 같이 dependency에 yield 사용가능
async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()

async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)

async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

 

 

# It might be tempting to raise an HTTPException or similar in the exit code, 
# after the yield. But it won't work.
# But if a background task creates a DB error, at least you can rollback or 
# cleanly close the session in the dependency with yield, and maybe log the 
# error or report it to a remote tracking system.