Fast API 배우기 19부 - Dependency Injection

Dependency Injection

Dependency Injection이란 코드의 재활용을 위해 제공해주는 fastapi의 기능이다.

 

가령 두개의 path operation이 있다고 하자

 

path1, path2 두개 모두 덧셈을 하는 로직이 필요하다.

그리고 두 path operation에 둘다 덧셈을 해주는 로직을 각각 작성하였다.

똑같은 로직인데 비효율적이지 않은가?

 

 

이에대한 해결책은 덧셈을 하는 로직을 하나 만들고 두개의 path operation에 dependency 를 걸어주면 되는것이다!

 

 

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

 

 

 

사용법은 아래와 같다.

 

재활용될 dependency 코드를 작성한다.

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

Depends를 import한다

from fastapi import Depends, FastAPI

dependency를 걸어준다

 

async def read_items(commons: dict = Depends(common_parameters)):
async def read_users(commons: dict = Depends(common_parameters)):

 

 

 

자 여기서 주목할 몇가지를 보자

 

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

 

디펜던시가 걸린 함수의 파라미터를 보면 path operation의 파라미터와 유사하다.

 

즉 dependency가 걸린 함수는 데코레이터(@app.get 같은거) 없는 path operation이라고 생각하면 된다!

 

다음은 Depends에 파라미터이다.

 

Depends에 올 파라미터는 함수처럼 사용이 가능한 파라미터가 와야하며, single 파라미터이여야 한다.

async def read_items(commons: dict = Depends(common_parameters)):
async def read_users(commons: dict = Depends(common_parameters)):

 

 

함수처럼 사용이 가능하다는 말은 꼭 함수가 아니여도 된다는것!

 

 

 

 

이제 docs를 확인해보자. 이전에 일반적으로 작성한 path operation 처럼 나온다!

 

 

Class Dependency

 

앞서 Depends의 파라미터는 함수처럼 쓸 수 있는 파라미터 하나가 와야한다고 했다.

 

그 말인 즉슨 클래스의 생성자는 함수처럼 사용이 가능하니 클래스도 가능하다는 얘기이다!

 

 

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]



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

 

 

클래스 디펜던시는 앞선 함수 파라미터에 비해 아래와 같은 장점을 가진다.

 

 

• 클래스 안에 여러가지 기능을 추가해서 리턴해 줄 수 있다
• 클래스의 멤벼변수가 key값이 되니 IDE에서 코드 작성할 때 자동완성에 이점이 있다

앞선 예제에서는 함수가 dict만 리턴해 주었다. 이를 사용할 때 q, skip, limit 같은 key값들은 ide에서 자동완성이 안되니 내가 수기로 입력해줘야 하는 불편함이 있다.

허나 클래스로 디펜던시를 사용하면 멤버변수이니 key값을 내가 찾지 않아도 자동완성 된다.

 

 

반응형

 

 

Depends의 Type annotation

 

허나 위 코드에 한가지 단점이 있다.

commons: CommonQueryParams = Depends(CommonQueryParams)

CommonQueryParams가 두번이나 반복되었다.

 

사실 fastapi는 = 뒤에있는 코드만 call을 한다.

... = Depends(CommonQueryParams)

 

그러므로 사실 아래처럼만 적어줘도 문제가 없다.

commons = Depends(CommonQueryParams)

 

 

그런데 한가지 문제가 있다.

 

commons의 타입을 명시해주지 않아서 IDE에서 자동완성이 안된다.

그래서 어쩔수 없이 개발의 편의성을 위해 아래처럼 적어주는게 낫다.

 

commons: CommonQueryParams = Depends(CommonQueryParams)

 

 

이문제를 어떻게 해결 할 수 있을가?

 

 

 

Shortcut

 

fastapi는 이를 해결해주기 위해 아래와 같은 shortcut을 제공한다.

 

commons: CommonQueryParams = Depends()

 

그러면 알아서 Depends가 CommonQueryParams 인스턴스를 만들어 리턴해준다.

 

그럼 코드중복도 해결하고 리턴타입이 명시되어있으니 IDE의 자동완성문제도 해결되었다!

 

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


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()):
    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

 

 

 

Sub Dependency

depency의 dependency를 만들 수 있다.

 

원하는 만큼 deep하게 만들수 있다.

 

from typing import Optional

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}

 

 

위코드를 설명하자면

 

디펜던시의 디펜던시를 만든다.

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

 

마지막으로 path operation에 연결시켜준다

async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):

 

 

Path Operation Decorator's Dependency

만약에 디펜던시가 리턴하는 값이 필요하지 않고 실행은 꼭 해야한다면

path operation decorator에 디펜던시를 달 수 있다.

 

사용방법은 데코레이터에 dependencies 파라미터에 Depends를 list형태로 넘겨준다.

 

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key



@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])

 

 

 

여기서 주목할점은 verify_key 디펜던시 이다. 값을 리턴하고 있다.

 

async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

하지만 저 리턴값은 path operation에 전달되지 않는다.

 

 

 

Global Dependency

개발을 하면서 어플리케이션 전체에 dependency를 걸어주고 싶다면 global dependency를 이용하면 된다

 

from fastapi import Depends, FastAPI, Header, HTTPException


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key



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



@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

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

사용법은 위와 같다.

 

 

 

 

Dependency with yield

FastAPI에서는 yield를 사용하여 디펜던시가 끝나면 몇가지 작업을 수행할 수 있다.

 

 

데이터베이스 디펜던시 with yield

대표적인 예제가, 데이터베이스를 생성한뒤 사용하고 사용이 끝났으면 데이터베이스를 종료하는 작업이 있다.

 

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

 

response를 보내기전에 yield 까지의 코드만이 실행된다.

 

그리고 response가 전달되면 마지막으로 finally 이후 코드가 수행된다.