퀴즈봇 만들기 - 3

folder 구성

 

app이라는 folder 안에는 config.py, datanase.py, main.py, models.py 이렇게 4개의 .py파일로 구성되어져 있고, lib에는 telegram.py와 telegram api를 활용하기 위한 .env 파일로 구성되어져 있습니다.

 

우선 제가 사용한 환경의 requirements.txt입니다.

altair==5.2.0
annotated-types==0.6.0
anyio==3.7.1
attrs==23.2.0
bcrypt==4.1.2
blinker==1.7.0
cachetools==5.3.3
certifi==2024.2.2
cffi==1.16.0
charset-normalizer==3.3.2
click==7.1.2
cryptography==3.4.7
defusedxml==0.7.1
ecdsa==0.18.0
exceptiongroup==1.2.0
fastapi==0.63.0
gitdb==4.0.11
GitPython==3.1.42
greenlet==3.0.3
h11==0.12.0
httpcore==0.13.7
httpie==3.2.2
httpx==0.18.1
idna==3.6
Jinja2==3.1.3
jsonschema==4.21.1
jsonschema-specifications==2023.12.1
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
multidict==6.0.5
numpy==1.26.4
packaging==23.2
pandas==2.2.1
pillow==10.2.0
protobuf==4.25.3
pyarrow==15.0.2
pyasn1==0.5.1
pycparser==2.22
pydantic==1.10.15
pydantic_core==2.16.3
pydeck==0.8.1b0
Pygments==2.17.2
PyMySQL==1.1.0
PySocks==1.7.1
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
python-jose==3.3.0
python-multipart==0.0.9
pytz==2024.1
referencing==0.34.0
requests==2.31.0
requests-toolbelt==1.0.0
rfc3986==1.5.0
rich==13.7.1
rpds-py==0.18.0
rsa==4.9
six==1.16.0
smmap==5.0.1
sniffio==1.3.1
SQLAlchemy==2.0.29
starlette==0.13.6
streamlit==1.32.2
tenacity==8.2.3
toml==0.10.2
toolz==0.12.1
tornado==6.4
typing_extensions==4.10.0
tzdata==2024.1
urllib3==2.2.1
uvicorn==0.13.4
watchdog==4.0.0

 

database.py의 코드 내용은 다음과 같습니다. 자세한 내용은 https://kbgw2001.tistory.com/55과 비슷합니다.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from app.config import settings

engine = create_engine(
    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}?charset=utf8mb4".format(
        username=settings.DB_USERNAME,
        password=settings.DB_PASSWORD.get_secret_value(),
        host=settings.DB_HOST,
        port=settings.DB_PORT,
        name=settings.DB_NAME,
    )
)

SessionLocal = sessionmaker(
    bind = engine,
    autocommit = False,
    autoflush=False
)

Base = declarative_base()

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

여기서 config.py 파일을 살펴보겠습니다 !! 

config.py의 구성은 다음과 같습니다.

from functools import lru_cache
from pydantic import SecretStr
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DB_USERNAME: str
    DB_PASSWORD: SecretStr
    DB_HOST: str
    DB_PORT: int
    DB_NAME: str # 

    TELEGRAM_BOT_TOKEN: SecretStr

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache
def get_settings():
    return Settings()

settings = get_settings()

config.py는 서버에 여러가지 값들을 설정을 할 때 사용을 하게 됩니다. 우선 pydantic에 BaseSettings를 import 해줬습니다. 그리고 가장 많이 쓰는 DB_USERNAME, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME을 config의 값으로 Setting했습니다.

그리고, DB_NAME은 DB에 DEV라는 database를 만들었습니다.

TELEGRAM_BOT_TOKEN이라는 값은 텔레그램에 BotFater에서 물어보면 됩니다.

이런식으로 API Token을 받을 수 있습니다.

 

대부분의 값들은 str로 되어져 있는데, DB_PASSWORD, TELEGRAM_BOT_TOKEN 2개가 SecretStr이라고 되어져 있는데 SecretStr은 print 문으로 출력을 해보았을 때, 실제 User가 입력한 비밀번호가 아닌 마스킹된 값을 보여주게됩니다 !! 

 

즉, 민감한 정보는 이와 같이 SecretStr을 사용합니다.

 

class Config:
    env_file = ".env"
    env_file_encoding = "utf-8"

그리고 인어 class인 이부분은 pydantic에 model을 설정을 해주는 부분입니다.

“.env”라는 파일이 있는데 이부분은 최상단에 .env라는 파일을 불러온다는 의미이고, encoding은 utf-8로하도록 설정을 했습니다.

 

.env 파일의 설정 값입니다. 이렇게 env 파일을 만들어주면 알아서 자동적으로 .env에서 값들을 가져오게 됩니다.

 

여기서 DB_USERNAME과 DB_PASSWORD와 DB_PORT는 퀴즈봇 만들기 - 1에서 설정한 DB 내용들입니다.

docker로 설정한 DB가 꺼져있다면 다음 명령어로 실행 시켜주면 됩니다.

sudo docker start [container name]
ex) sudo docker start test-db

 

Settings을 바로 instance화 해주는 것이 아니라 함수를 통해서 instance화 하도록 하는 부분입니다.

@lru_cache
def get_settings():
    return Settings()

settings = get_settings()

 

 

lru_cache라는 데코레이션을 사용을하는데, lru_cache는 캐싱을 해주는 데코레이션입니다. Settings의 경우, 자주 불러와 사용을 하기 때문에 이렇게 캐싱을 해줘야 더 빠르게 접근이 가능하게 됩니다. lru_cache는 최근에 많이 사용하는 캐싱 알고리즘이라고 합니다. 

다음은 models.py입니다. models.py는 다음과 같이 setting을 했습니다

from sqlalchemy import Column, Integer, String
from app.database import Base

class User(Base):
    __tablename__="user"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    password = Column(String(255))

 

우선을 이렇게 User의 정보를 받도록 했습니다. main.py의 코드도 살펴 보겠습니다.

from fastapi import FastAPI
from 
from app import models
from app.config import settings
from lib.telegram import Telegram

app = FastAPI()
telegram = Telegram(settings.TELEGRAM_BOT_TOKEN)

@app.on_event("startup")
def on_startup():
    for app.database import engine
    models.Base.metadata.create_all(bind = engine)

@app.get("/")
async def hello():
    return b"hello world!"

@app.get("/me")
async def get_me():
    return await telegram.get_bot_info()

 

여기서 우선 이 부분을 살펴 보겠습니다.

@app.on_event("startup")
def on_startup():
    for app.database import engine
    models.Base.metadata.create_all(bind = engine)

 

@app.on_envent(”startup”)은 FastAPI가 실행이되면, 아래있는 코드를 실행하겠다라는 의미입니다.

 

models.Base.metadata.create_all이라는 함수를 사용해서 engine에 있는 class들을 전부 DB에 Table로 생성을 하겠다라는 의미입니다.

main.py를 uvicorn을 사용해서 커멘드라인에서 실행해 보겠습니다.

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

 

위의 main.py에 me라는 domain을 지정을 했습니다.

@app.get("/me")
async def get_me():
    return await telegram.get_bot_info()

 

HTTPie를 사용해서 해당 부분에 대한 호출을 해보겠습니다.

http :8000/me

 

다음처럼 텔레그램에서 setting 했던 bot에 대한 내용들이 등장을 합니다.

 

lib 바로 아래 이렇게 telegram.py라는 파일이 있습니다.

from typing import Union

import httpx
from pydantic import SecretStr


class Telegram:
    API_HOST = "https://api.telegram.org"

    def __init__(self, token: Union[str, SecretStr]) -> None:
        self._token = token
        self.client = httpx.AsyncClient(base_url=self.host)

    @property
    def token(self):
        if isinstance(self._token, SecretStr):
            return self._token.get_secret_value()
        return self._token

    @property
    def host(self):
        return f"{self.API_HOST}/bot{self.token}/"

    async def get_bot_info(self) -> dict:
        """Get current bot info"""
        r = await self.client.get("getMe")
        return r.json()

 

보통 API에서 다 처리가 가능하지만, 이렇게 작성하는 이유는 API과 Code가 따로 동작이 되기 때문에 유지 보수 하기가 좋다는 장점이 있습니다.

간단하게 살펴 보겠습니다.

API_HOST = "https://api.telegram.org"

telgram의 API_HOST는 다음과 같이 되어져 있고, 해당 주소는 텔레그램의 실제 API의 Host 주소입니다.

https://core.telegram.org/bots/api 해당 사이트에 접속을 한 후에 Making requests 부분을 살펴보면,

https://api.telegram.org/bot/METHOD_NAME

다음과 같이 API_HOST와 bot token과 METHOD_NAME을 붙여서 요청을 보내라고 합니다. 

이 부분을 구현한 것이 Telegram class입니다.

 def __init__(self, token: Union[str, SecretStr]) -> None:
        self._token = token
        self.client = httpx.AsyncClient(base_url=self.host)

init 부분을 보면, token을 입력을 받아서 self._token = token에 저장을 합니다.

token의 경우 불변하는 값이기 때문에 다음과 같이 property로 저장을 했습니다.

@property
def token(self):
    if isinstance(self._token, SecretStr):
        return self._token.get_secret_value()
    return self._token

config.py에서 token을 secretstr로 지정을 해두었기 때문에 그냥 읽게 되면 마스킹 된 값이 들어가기 때문에 token이 SecretStr의 경우에는 get_secret_value() 함수를 통해 원래 token값을 return하도록 했습니다.

만약 SecretStr이 아닌 경우에는 token을 return 하도록 했습니다.

 def __init__(self, token: Union[str, SecretStr]) -> None:
        self._token = token
        self.client = httpx.AsyncClient(base_url=self.host)

init 부분에서  httpx의 어싱크 클라이언트 함수를 사용해서 클라이언트를 만들어 줍니다.

base_url에 들어가는 값은 self.host라는 값이 들어가는 값은 property로 설정된 host 함수의 값이 들어가게 됩니다.

@property
 def host(self):
      return f"{self.API_HOST}/bot{self.token}/"

 

해당 부분이 아래 양식에 해당하는 부분입니다.

https://api.telegram.org/bot/METHOD_NAME

host 함수에서 return된 값을 base_url로 지정을 하게됩니다. 이렇게 하게 되면, get_bot_info를 사용해서 get에 Method 명만 입력을 한다면, 바로 API를 호출할 수 있게됩니다.

main.py 부분에 

@app.get("/me")
async def get_me():
    return await telegram.get_bot_info()

 

me라 path로 이동하게 되면, getMe라는 Method를 사용한 정보를 return하게 됩니다.

 

https://core.telegram.org/bots/api에서 검색을 해보면, 다음과 같이

 

 

봇에 대한 기본 정보를 반환하는 method라고 합니다. 위의 코드 통해서 편하게 실제 API의 method를 호출했다고 생각하면됩니다.

 

telegram bot을 만들고 간단하게 API를 호출해보고  test하는 코드를 작성해 보았습니다 !!