이 시리즈를 통해 FastAPI 프로젝트를 배포했습니다. 하지만 서비스가 업데이트될 때마다 이미지를 ECR에 업로드하고 ECS 작업 정의를 생성한 뒤 업데이트를 해야하는 과정은 꽤나 번거롭습니다.

이번 포스팅에선 이 번거로운 작업들을 자동화하기 위해 CI/CD를 구축하는 방법을 살펴보겠습니다.

2. Github Repository 생성

먼저 github에세 새로운 레포지토리를 생성합니다. 이후 "Setting"에 들어와서 왼쪽 "Security" 패널의 "Secrets and variables"를 클릭하여 "New repository secret"을 생성합니다.

여기서 AWS IAM 사용자 생성을 했을 때 발급 받은 액세스 키로 아래와 같이 Name과 Secret을 설정하고 환경변수를 각각 생성합니다.

• AWS_ACCESS_KEY_ID : 발급받은 aws access key
• AWS_SECRET_ACCESS_KEY : 발급받은 aws secret access key

환경변수가 생성되었다면 "Actions" 탭에서 Deploy to Amazon ECs를 검색해 선택합니다.

이제 aws.yml의 env에서 아래의 항목을 수정해야합니다.

• AWS_REGION: your aws region
• ECR_REPOSITORY: AWS ECR Repository name
• ECS_SERVICE: AWS ECS Service name
• ECS_CLUSTER: AWS ECS Cluster name
• ECS_TASK_DEFINITION: .aws/task-definition.json

설정이 되었다면 프로젝트 루트 디렉토리에 .aws 폴더를 생성하고 그 안에 task-definition.json 파일을 생성합니다.

이후 AWS ECS에서 테스크 정의에 생성된 테스크 정의의 JSON 내용을 복사해 task-definition.json에 붙여넣기 합니다.

이제 다시 github "Actions"으로 돌아와 "Commit"을 눌러줍니다.

3. CI/CD

이 aws.yml 파일은 새 버전의 서비스를 AWS ECS에 배포하기 위해 수행해야 하는 작업을 정의합니다. 사전에 정의된 동작은 main branch로 push할 때마다 github action이 배포를 시작하는 것입니다.

이제 git을 초기화하고 프로젝트 디렉토리와 연결해 push 작업을 진행해야합니다. git clone으로 해당 레포지토리를 가져온 후 프로젝트를 해당 레포지토리에 복사한 후 push합니다.

git clone <your repository url> <dir name>

git add .
git commit -m "first commit"
git push -u origin main

그리고 "Actions" 탭으로 이동하면 워크플로우의 단계별 절차를 확인할 수 있습니다.

잠시 후 배포가 완료된 것을 확인할 수 있습니다.

4. 테스트

이제 코드를 수정해 github action을 통해서 재배포를 해보겠습니다.

import os
from fastapi import FastAPI

app = FastAPI()

@app.get('/')
def welcome_root():
    my_name = os.environ.get("MY_NAME")
    return f"Hello {my_name}!"

변경사항을 수정 후 main 브랜치에 푸시합니다.

git add .
git commit -m "update code"
git push

Github "Actions" 탭으로 이동해 새배포가 완료되었으면 브라우저를 열어 확인합니다.

이전 포스팅에선 FastAPI 서버의 컨테이너 이미지를 AWS ECR에 업로드했습니다.

이제 서버를 ECS Fargate에 배포해보도록 하겠습니다.

2. ALB(Application Load Balancer) 생성

AWS EC2로 이동해 "로드 밸런싱"의 "로드밸러서"로 이동해 로드밸러스를 생성합니다.

로드 밸런서 유형은 "Application Load Balancer"로 생성합니다.

로드 밸런서의 이름을 설정하고 네트워크 매핑에서 VPC를 선택합니다. 만약 VPC가 없다면 VPC로 이동해서 VPC를 생성해줍니다.

최소 2개의 가용영역과 영역당 하나의 서브넷이 필요합니다.

새로운 창을 열어 AWS EC2 보안그룹에서 HTTP 요청에 80 포트를 허용하는 보안그룹을 생성합니다. 생성한 VPC를 선택하고 인바운드 규칙에서 아래와 같이 설정합니다.

보안그룹에서 새로고침을 눌러 생성한 보안그룹을 적용합니다.

이제 타켓이 되는 그룹을 선택해야합니다. AWS EC2 대상 그룹 으로 이동해 대상 그룹 생성을 합니다.

로드 밸런싱을 지원하는 "IP 주소"를 선택하고 생성한 VPC를 선택 후 다음과 같이 설정합니다.

다시 로드 밸런서로 돌아와 리스너 및 라우팅에서 새로고침을 누르고 생성한 대상 그룹을 선택 후 로드 밸런서를 생성합니다.

3. ECS(Elastic Container Service)

이제 AWS ECS 로 이동해 클러스터를 생성합니다. AWS Fargate로 배포할 예정이니 인프라를 AWS Fargate로 설정합니다.

AWS ECS의 테스크 정의로 이동해 새 테스크 정의 생성을 합니다. 테스크 정의는 포트 매핑, 도커 이미지, CPU, 메모리 사용량 등과 같은 서비스의 동작 및 설정을 정의하는 docker-compose과 유사합니다.

AWS Fargate를 선택하고 각자에 필요에 따라 CPU와 메모리 사용량을 설정하고 테스크 역할을 "ecsTaskExecutionRole"로 선택합니다.

이제 컨테이너 포트를 매핑해야합니다. 이전에 생성했던 컨테이너 이미지의 이름과 URL을 ECR Repogitories에서 가져옵니다.

아래와 같이 이미지의 이름과 URL을 설정합니다. 또한 컨테이너는 8000포트를 수신하기 때문에 컨테이너 포트가 8000임을 나타내는 포트 매핑을 정의합니다.

이후 테스크를 생성하고 다시 클러스터로 이동해 생성한 클러스터에서 서비스를 생성합니다.

컴퓨팅 옵션을 시작 유형으로 선택합니다.

패밀리를 이전에 생성한 테스크 정의를 선택하고 아래와 같이 배포구성을 합니다.

아래로 이동해 네트워킹 설정을 합니다. 생성한 VPC와 서브넷을 선택한 후 아래와 같이 보안그룹을 생성합니다. 이전에 컨테이너를 8000 포트에 매핑했으므로 사용자 지정으로 TCP 8000 포트를 허용하는 보안 그룹을 생성해야합니다.

이제 로드 밸런싱으로 이동해 생성한 로드 밸랜싱을 연결합니다. 기존 로드 밸런서에서 생성한 로드 밸런서를 선택한 후 생성한 대상 그룹을 적용합니다.

서비스 성공적으로 배포되면 아래와 같이 활성화 됩니다.

이제 테스트를 위해 EC2의 로드밸런서로 이동해 DNS 이름으로 브라우저에서 접근합니다.

짠! 제대로 동작하는 것을 확인할 수 있습니다.

Pyhton에서 코드를 구현하다보면 여러 에러를 맞이하게 됩니다. 이렇듯 코드를 실행하는 중에 발생한 에러는 exception(예외)라 합니다.

예외처리를 하기위해 python에서는 try except 구문이 존재합니다. try에 실행할 코드를 구현하고 except에 예외가 발생했을 때 처리하는 코드를 구현합니다.

이 포스팅에선 try except에 대해 살펴보고 사용자 정의 예외를 구현하고 이를 FastAPI에서 적용시켜보고 저의 Jobdam에 어떻게 적용되었는지 확인해보겠습니다!

Handling Exceptions

에러는 문장이나 표현식이 올바르다 할지라도 실행할 때 에어를 일으킬 수 있습니다. 실행 중에 감지되는 에러들을 예외(exception)라 부르고 이는 무조건 치명적이지는 않습니다. 아래는 올바른 표현식이나 프로그램이 처리하지 않은 에러의 예시입니다.

print(10 * (1/0))

Traceback (most recent call last):
  File "C:\Users\cream\desktop\reference\main.py", line 1, in <module>
    print(10 * (1/0))
                ~^~
ZeroDivisionError: division by zero

에러의 마지막줄의 ZeroDivisionError 는 내장 예외 중 하나입니다. 에러의 나머지는 예외의 형태와 원인에 기반을 둔 상세 내용을 제공합니다.

이러한 예외들을 try except 구문을 통해 선택적으로 처리할 수 있습니다. 아래는 올바른 정수가 입력될 때까지 예외를 일으키는 예시입니다.

while True:
    try:
        x = int(input("Please enter a number: "))
        break
    except ValueError:
        print("Oops!  That was no valid number.  Try again...")

• 먼저 try 바로 아래의 코드가 실행됩니다.
• 예외가 발생하지 않으면 except를 건너뛰고 try 문의 실행은 종료됩니다.
• 예외가 발생하면 try 구문의 나머지를 건너뛰고 except 에서 해당 유형의 에러가 발생하면 예외처리가 됩니다.
• 만약 except에서 제시된 유형의 에러가 발생하지않으면 에러 메시지와 함께 실행이 종료됩니다.

BaseException의 하위 클래스 중 하나인 Exception은 치명적이지 않은 모든 예외의 기본 클래스입니다. (예외 계층 은 여기서 확인 가능합니다.) 만약 위의 예시에서 무슨 에러가 발생하는지 잘 모르겠다면 Exception을 통해 거의 모든 예외를 처리할 수 있습니다.

while True:
    try:
        x = int(input("Please enter a number: "))
        break
    except Exception:
        print("Oops!  That was no valid number.  Try again...")

하지만 처리하려는 예외 유형을 최대한 구체적으로 지정하는 것이 좋습니다. 만약 더 정확한 에러를 알고 싶다면 except 구문에서 as 뒤에 변수를 지정해서 에러 메시지를 확인할 수 있습니다.

while True:
    try:
        x = int(input("Please enter a number: "))
        break
    except Exception as e:
        print("Oops!  That was no valid number.  Try again...")
        print(e)

Please enter a number: s
Oops!  That was no valid number.  Try again...
invalid literal for int() with base 10: 's'

Raising Exceptions

raise 구문은 사용자가 직접 지정한 예외가 발생하도록 강제할 수 있습니다. raise는 예외 인스턴스거나 예외 클래스(BaseException 하위 클래스) 중 하나 이어야 합니다.

아래는 raise 구문으로 직접 예외를 발생하도록 처리하고 try except 에서 처리한 예외를 다시 발생시키는 예시입니다.

def test_exception():
    try:
        x = int(input("Please enter a number: "))
        if x % 2 != 0:
            raise Exception('Oops!  That was no valid number')
        print(x)
    except Exception as e:
        print('Exception in function', e)
        raise

try:
    test_exception()
except Exception as e:
    print('Exception in parent code', e)

Please enter a number: ;
Exception in function invalid literal for int() with base 10: ';'
Exception in parent code invalid literal for int() with base 10: ';'

FastAPI Handling Errors & Custom exception handlers

FastAPI에선 일반적으로 HTTPException을 이용해 status_code와 에러 메시지인 detail로 예외처리를 합니다.

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

클라이언트가 정상 요청을 할 경우 HTTP 상태코드인 200과 함께 다음 JSON 응답을 받게 됩니다.

{
  "item": "The Foo Wrestlers"
}

비정상 요청인 경우 HTTP 상태코드인 404와 함께 다음 JSON 응답을 받게 됩니다.

{
  "detail": "Item not found"
}

이제 간단한 사용자 정의 예외를 만들어 이 예외를 전역적으로 처리하는 코드를 작성해보겠습니다. UnicornException 이라는 사용자 정의 예외를 생성하고 @app.exception_handler()를 통해 사용자 정의 예외를 추가했습니다.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


app = FastAPI()


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

/unicorns/yolo로 요청하면 raise UnicornException 예외가 발생하고 이는 unicorn_exception_handler 에서 처리하고 다음과 같은 응답을 받게 됩니다.

{"message": "Oops! yolo did something. There goes a rainbow..."}

User defined Exceptions with FastAPI

이것만으로도 충분한 사용자 예외 처리를 할 수 있어보입니다. 하지만 프로젝트를 진행하다보면 수많은 예외를 발생시켜야 할 때가 많습니다. 그렇기에 메인이 되는 BaseException 부모 클래스를 생성하고 이를 상속받아 FastAPI에서 exception_handler로 정의하면 재활용이 쉽고, 손쉽게 예외 처리를 할 수 있습니다.

저는 FastAPI에서 회원가입이 되지 않은 회원이 로그인을 시도하려할 때 발생시킬 사용자 정의 예외를 만드려고 합니다.

아래와 같이 Exception 클래스를 상속받아 사용자 정의 예외를 만들고 python 내장 클래스인 __str__을 정의해 객체가 print 함수에 전달될 경우 객체 내부의 detail을 반환하도록 했습니다.

class BaseCustomException(Exception):
    """Base class for custom exceptions"""

    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail

    def __str__(self):
        return self.detail

직접 정의한 BaseCustomException을 상속받아 UnregisteredUserError 예외 클래스를 생성합니다. 여기선 super().__init__ 을 통해 부모 클래스인 BaseCutomException의 인스턴스를 가져옵니다.

...

class UnregisteredUserError(BaseCustomException):
    def __init__(self, user_name: str):
        super().__init__(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"{user_name} is an unregistered user."
        )

그리고 직접 정의한 BaseCustomException을 FastAPI에서 exception_handler로 사용하기 위한 handler를 생성합니다.

...

def base_custom_exception_handler(request: Request, exc: BaseCustomException):
    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})

이를 add_exception_handler 를 통해 예외 핸들러를 추가해줍니다.

...
app = FastAPI()

...
app.add_exception_handler(BaseCustomException, base_custom_exception_handler)

이것이 실제로 적용된 코드가 궁금하다면 Jobdamserver에서 확인이 가능합니다!

reference

• Python 공식문서 exceptions
• FastAPI 공식문서 handling error

Jobdam 프로젝트에서 Python을 사용하여 HTTP 통신을 구현해야 했습니다.

저는 Python 기반의 TUI(Terminal User Interface)인 Textual로 터미널 앱 Jobdam을 개발한 경험이 있습니다. 또한, 해당 프로젝트의 서버는 Python 기반의 FastAPI를 사용하여 구축되었기 때문에 Python을 이용한 HTTP 통신이 필수적이었습니다.

이를 위해 python으로 HTTP 통신을 가능하게 해주는 requests라이브러리를 사용하게 되었습니다. 이번 포스팅에서는 requests 라이브러리에 대한 개요와 Jobdam 프로젝트에서의 적용 방법에 대해 알아보겠습니다.

2. 환경설정 및 Requests 라이브러리 개요

Requests 라이브러리는 HTTP 통신을 위한 다양한 메서드를 제공합니다. 이를 이용하여 REST API 방식의 Web API를 호출하고 데이터를 요청, 수정할 수 있습니다. 주요 메서드로는 get, post, patch, put, delete 등이 있습니다.

먼저 가상환경을 생성하고 패키지를 설치해줍니다.

python -m venv venv

source venv/scripts/activate

pip install requests

3. Requests method with FastAPI

FastAPI는 빠르고 현대적인 Python 웹 프레임워크로, requests 라이브러리와 함께 사용하면 효율적인 웹 개발이 가능합니다.

이를 간단한 FastAPI 서버를 만들어서 테스트 해보겠습니다.

FastAPI 패키지를 설치해줍니다.

pip install "fastapi[all]"

FastAPI를 이용하여 간단한 서버를 만들어보겠습니다. server.py 파일을 생성하고 다음의 코드를 작성합니다.

3-1. GET

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

이제 클라이언트에서 해당 서버에 요청을 보내보겠습니다. client.py 파일을 생성하고 아래의 코드를 작성합니다.

import requests

url = "http://localhost:8000"

resp = requests.get(url=url)

print(f"status_code: {resp.status_code}")
print(f"content: {resp.content}")
print(f"headers: {resp.headers}")

출력결과는 아래와 같습니다.

• resp.status_code는 말 그대로 HTTP 상태 코드를 반환합니다
• resp.content는 바이너리 데이터 값을 반환합니다.
• resp.headers는 response의 메타 데이터를 반환합니다.

status_code: 200
content: b'{"Hello":"World"}'
headers: {'date': 'Fri, 08 Dec 2023 13:09:06 GMT', 'server': 'uvicorn', 'content-length': '17', 'content-type': 'application/json'}

또한 get 방식으로 HTTP 요청을 할 경우 query string을 통해 응답받을 데이터를 필터링하는 경우가 많습니다.

이는 get의 params 인자로 넘겨주면 query string을 지정할 수 있습니다.

import requests

url = f"http://localhost:8000/items/1"

resp = requests.get(url=url, params={"q": 1})

print(f"status_code: {resp.status_code}")
print(f"content: {resp.content}")
print(f"headers: {resp.headers}")

status_code: 200
content: b'{"item_id":1,"q":"1"}'
headers: {'date': 'Fri, 08 Dec 2023 13:16:03 GMT', 'server': 'uvicorn', 'content-length': '21', 'content-type': 'application/json'}

만약 params를 쓰지 않는다면 url에 그대로 모든 query string을 넣어줘도 무관합니다.

...
url = f"http://localhost:8000/items/1?q=1"
...

3-2. POST

이제 HTTP 요청에 데이터를 담아서 보내는 방식인 post에 대해서 살펴보겠습니다. Requests의 post는 data와 json 옵션으로 데이터를 담아서 보낼 수 있습니다.

data 옵션은 HTML 양식(form) 포맷의 데이터를 전송할 수 있고 content-type 요청 헤더는 application/x-www-form-urlencoded로 설정됩니다. json 옵션은 REST API 형식으로 JSON 포맷의 데이터를 전송할 수 있으며 content-type 요청 헤더는 application/json으로 설정됩니다.

server.py 를 아래와 같이 변경해줍니다. 2개 이상의 데이터를 받기 위해 pydantic의 BaseModel을 상속한 클래스를 생성해줍니다.

from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()


class UserData(BaseModel):
    user_name: str
    password: str


@app.post("/login")
def login(user_data: UserData):
    results = {"user_data": user_data}
    return results

이후 client.py 코드를 아래와 같이 변경 후 data 옵션을 이용해 요청해보겠습니다.

import requests

url = f"http://localhost:8000/login"

data = {
    "user_name": "earthquake",
    "password": "pass123word123"
}

resp = requests.post(url=url, data=data)

print(f"status_code: {resp.status_code}")
print(f"content: {resp.content}")
print(f"headers: {resp.headers}")

data 옵션을 이용해 요청했더니 아래와 같은 에러를 발생시킵니다. 이유는 FastAPI는 별다른 양식 필드(form)을 지정하지 않으면 JSON으로 요청을 해야하기 때문입니다.

status_code: 422
content: b'{"detail":[{"type":"model_attributes_type","loc":["body"],"msg":"Input should be a valid dictionary or object to 
extract fields from","input":"user_name=earthquake&password=pass123word123","url":"https://errors.pydantic.dev/2.5/v/model_attributes_type"}]}'
headers: {'date': 'Fri, 08 Dec 2023 13:34:56 GMT', 'server': 'uvicorn', 'content-length': '255', 'content-type': 'application/json'}

옵션을 다시 json으로 변경한 뒤 실행하면 정상적인 응답이 오는 것을 확인할 수 있습니다. json 옵션을 이용하면 dictionary로 들어온 객체도 JSON의 형태로 변환 후 요청이 됩니다.

...
resp = requests.post(url=url, json=data)
...

status_code: 200
content: b'{"user_data":{"user_name":"earthquake","password":"pass123word123"}}'
headers: {'date': 'Fri, 08 Dec 2023 13:35:05 GMT', 'server': 'uvicorn', 'content-length': '68', 'content-type': 'application/json'}

4. 프로젝트 적용 사례

아래는 Jobdam프로젝트에서 requests 모듈을 이용해 HTTP 통신을 적용한 사례입니다. create_chat_room 이라는 함수에 dictionary 데이터를 인자로 받아 해당 endpoint에 데이터를 전송하고 지정된 리턴값을 받습니다.

def create_chat_room(self, data: dict):
    resp = requests.post(
        url=cfg.base_url + "/chat_room/create",
        json=data,
        headers=auth_utils.build_jwt_header(cfg.config_path)
    )
    if resp.status_code == 201:
        create_room_resp = global_utils.bytes2dict(resp.content)
        return {"status_code": resp.status_code}
    elif resp.status_code == 400:
        detail = global_utils.bytes2dict(resp.content)['detail']
        return {"status_code": resp.status_code, "detail": detail}
    else:
        detail = global_utils.bytes2dict(resp.content)['detail']
        return {"status_code": resp.status_code, "detail": detail}

여기서 주목해야할 점은 5번째 줄에 headers입니다.

제가 만든 app에서는 해당 요청을 보내는 사용자가 인증이 된 사용자인지를 확인해야합니다. 그러기 위해선 HTTP Authorization request header를 post 요청의 headers에 실어서 보내야합니다. HTTP Authorization request header의 양식은 다음과 같습니다.

Authorization: <type> <credentials>

• type은 인증 타입 혹은 인증 스키마라고도 불리며 대표적인 예로는 Bearer 와 jwt가 있습니다.
• credentials는 base64 형태로 인코딩되는 것으로 흔히 불리는 access token이 이에 해당합니다.

auth_utils.build_jwt_header(cfg.config_path)의 함수를 보면 HTTP Authorization request header의 양식을 반환하는 것을 알 수 있습니다.

def build_jwt_header(fpath):
    return {
        "Authorization": "Bearer " + get_access_token_from_json(fpath)
    }

이를 통해 요청을 하는 사용자가 인증된 사용자라면 정상적인 응답을 받을 수 있고 만약 인증된 사용자가 아니라면 아래와 같이 인증되지 않았다라는 응답을 받게 됩니다.

"GET /chat_room/all_rooms_list HTTP/1.1" 401 Unauthorized

또한 8번째 줄을 보면 resp.content 를 global_utils.byes2dict() 라는 함수에 인자로 넘겨준 것을 확인할 수 있습니다.

global_utils.byes2dict 함수는 아래와 같이 바이너리 형태로 넘어오는 resp.content를 dictionary 객체로 변환해주는 역할을 합니다.

def bytes2dict(b):
    return json.loads(b.decode('utf-8'))

requests 모듈을 직접 적용한 프로젝트의 코드를 확인하고 싶다면 Jobdam에 방문해주세요!

reference

• Requests 라이브러리 공식 문서
• 파이썬에서 requests 라이브러리로 원격 API 호출하기

이전 포스팅에선 docker를 통해 이미지를 빌드했습니다. 이번에는 Docker hub와 같은 컨테이너 임지에 대한 레지스트리인 AWS ECR(Elastic Container Registry)에 컨테이너를 업로드하는 과정을 살펴보겠습니다.

2. IAM 사용자 생성

컨테이너 이미지를 ECR 저장소에 게시하기 위해선 먼저 IAM(Identity and Access Management)에서 사용자를 생성해야합니다.

AWS IAM 에 접속해서 "액세스관리"의 사용자를 눌러 사용자 생성을 해줍니다.

사용자 이름을 입력하고 다음을 눌러 "직접 정책 연결"을 선택한 후 다음과 같이AmazonEC2ContainerRegistryFullAccess 와 AmazonECS_FullAccess 정책을 추가합니다

이후 생성을 눌러 사용자를 생성하고 생성된 사용자를 클릭해 "보안 자격 증명"을 클릭합니다. 아래의 액세스 키 생성을 눌러 "Command Line Interface(CLI)"로 액세스 키를 생성합니다.

생성된 액세스 키는 CSV 파일로 다운하거나 자신의 프라이빗 저장소에 저장합니다.

ECR repository 생성

Amazon ECR로 접속해 리포지토리 생성을 클릭합니다.

다음과 같이 리포지토리의 이름을 정해 리포지토리를 생성합니다.

생성한 리포지토리에 들어와 "푸시 명령 보기"를 클릭하면 명령어들이 제시되는 것을 볼 수 있습니다.

4. 도커 이미지 ECR repository에 업로드

액세스 키의 접근 방식을 Command Line Interface(CLI)로 생성했기 때문에 aws-cli 설치를 필요로 합니다. AWS-CLI 해당 페이지에서 aws-cli를 다운로드 후 설정을 진행합니다.

터미널에서 아래의 설정해줍니다.

$ aws configure

-> AWS Access Key ID [None]: <Acess Key id>
-> AWS Secret Access Key [None]: <Secret Access Key>
-> Default region name [None]: <Your region name>
-> Default output format [None]:

이후 위에서 제시된 푸시 명령들을 순서대로 하나씩 진행합니다.(도커가 실행되고 있는 상태여야합니다.)

마지막 명령어를 입력하면 ECR repository에 업로드가 끝났다면 새로고침을 버튼을 눌러 업로드된 이미지를 확인합니다.

이 시리즈에선 제가 Jobdam이란 프로젝트에서 서버를 실제로 어떻게 배포했는지 살펴보겠습니다.

프로젝트를 배포하는 한 가지 방법은 Docker를 사용하여 컨테이너 이미지를 빌드하는 것입니다. 이 글에서는 Docker의 기본 개념과 함께 FastAPI 프로젝트를 어떻게 컨테이너 이미지로 만드는지 알아봅니다.

2. Docker(도커)란?

도커는 소프트웨어를 컨테이너(Container) 라는 표준화된 유닛으로 패키징하여 애플리케이션을 구축하고 배포할 수 있게 해주는 오픈소스 프로젝트입니다.

간단하게 말하면, 애플리케이션의 코드와 설치된 라이브러리 등을 컨테이너라는 공간에 담아서 배포하기 쉽게 도와주는 플랫폼입니다.

그렇다면 컨테이너란 무엇일까요

2-1. 컨테이너(Container)

흔히 말하는 소프트웨어는 OS와 라이브러리에 의존성을 가집니다. 만약 서로 다른 OS나 버전이 다른 라이브러리의 소프트웨어를 실행하고자 할 때 오류를 가져올 수 있고 이를 제어하기가 어려워집니다.

컨테이너는 소프트웨어(애플리케이션)를 실행할 때 독립적인 환경을 제공해주는 운영체제 수준의 격리 기술을 말합니다.

설명을 위해 가상 머신과 컨테이너를 비교해보겠습니다. 아래 사진과 같이 가상 머신은 OS 레벨을 포함한 전체를 가상화하는 가상머신과 달리 격리된 프로세스로 실행되는 컨테이너는 가상머신에 비해 매우 가볍습니다.

쉽게 얘기하면 window에서 개발하는 사람과 mac에서 개발하는 사람은 OS가 다릅니다. 또한 사용하는 언어의 버전이나 라이브러리 버전 등이 다를 수 있는 것을 방지하고자 컨테이너를 사용하여 동일한 OS단에서 코드와 라이브러리 버전의 통일성을 만드는 것입니다.

여기서 컨테이너 실행에 필요한 프로그램과 소스코드, 라이브러리 등을 포함시킨 것 파일을 Docker image(도커 이미지)라고 합니다.

2-2. 도커 이미지(Docker image)

도커 이미지는 서비스 운영에 필요한 프로그램, 소스코드, 라이브러리 등을 포함한 파일입니다. 즉, 특정 프로세스가 실행되기 위한(여기선 컨테이너 실행) 모든 파일과 환경을 지닌 것으로 더 이상의 의존성 파일을 컴파일하거나 라이브러리를 설치할 필요가 없는 상태의 파일을 의미합니다.

이미지는 읽기 전용(read-only)으로 스냅샷이라고도 합니다. 따라서 이미지는 실행할 수 없는 정적 이미지(불변성) 입니다. 즉, 이미지는 빌드 타임 구조로 이루어져 있고 컨테이너는 런타임 구조로 실행 중인 이미지를 나타냅니다.(예를 들면 객체 지향 언어에서의 클래스와 해당 클래스 객체의 차이인 것 같습니다?)

2-3. 레이어(Layer)

도커 이미지는 컨테이너를 생성하기 위한 모든 정보를 가지고 있기 때문에 보통 수백MB ~ 수GB가 넘는다. 그런데 기존 이미지에서 작은 변경사항이 생겨 도커 파일에 코드 몇 줄을 추가해 다시 이미지를 만들고 다시 그 이미지를 다운 받는다고 가정합니다.

이때 이미지의 불변성 때문에 수백MB ~ 수백GB가 되는 이미지를 다시 다운로드 받는 것은 매우 비효율적입니다. 도커는 이러한 문제를 해결하기 위해 레이어(Layer)라는 개념을 도입했습니다.

레이어란 수정사항이나 추가적인 파일이 있을 때 다시 다운로드 받는 것이 아닌 파일을 추가하는 개념입니다. 아래의 사진과 같이 기존의 이미지가 존재하는데 새로운 이미지를 다운 받을 경우 레이어만 따로 빼 파일을 추가하는 형식입니다.

도커 이미지는 위 그림처럼 여러 개의 읽기 전요(read-only) 레이어로 구성되고, 파일이 추가되면 새로운 레이어가 생성됩니다. 이러한 개념은 git 레포지토리에 commit 로그를 쌓는 것과 같다고 볼 수 있습니다.

3. 프로젝트 생성

이제 이론을 바탕으로 실전에 적용해볼 차례입니다. 간단한 FastAPI 프로젝트를 생성하고 docker를 이용하여 컨테이너 이미지를 빌드해보겠습니다.

새로운 폴더를 만들고 가상환경 생성 후 활성화합니다.

python -m venv venv

source venv/scripts/activate

이후 필요한 라이브러리를 설치하고 종속성을 파일에 저장합니다.

pip install "fastapi[all]"

pip freeze > requirements.txt

app 폴더를 생성 후 해당 폴더에 main.py 파일을 만들고 루트 도메인에 간단한 문구를 출력하는 FastAPI 서버 코드를 작성해보겠습니다.

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
def welcome_root():
    return "Welcome to root"

다음과 같은 명령어로 실행하면 서버가 잘 동작하는 것을 확인할 수 있습니다.

uvicorn app.main:app --reload

4. Dockerfile 생성

루트 디렉토리에 Dockerfile을 생성하고 다음과 같이 입력합니다.

FROM python:3.11

WORKDIR /test

COPY ./requirements.txt /test/requirements.txt

RUN pip install --no-cache-dir --upgrade -r /test/requirements.txt

COPY ./app /test/app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

여기서 중요한 것은 다음과 같은 코드입니다.

COPY ./requirements.txt /test/requirements.txt

먼저 코드의 나머지 부분이 아닌 종속성을 모아둔 파일을 dockerfile에 복사합니다. 이것은 위에선 언급했던 레이어 개념이 적용된 부분입니다.

도커는 컨테이너 이미지를 점진적으로 구축하여 수정사항이나 새로운 파일이 추가되면 기존의 레이어에 추가되어 구축됩니다. 이때 도커가 수정되지 않은 레이어를 발견하면 캐시에서 해당 레이어를 재사용하여 시간을 절약할 수 있습니다.

즉, 종속성을 저장하는 파일인 requirements.txt는 소스코드에 비해 자주 수정되는 파일이 아니기 때문에 소스 파일 위에 종속성 파일을 설치하는 명령어를 배치하는 것이 더 효율적입니다.

모든 작업이 끝났다면 프로젝트 구조는 다음과 같습니다.

.
├── app
│   ├── __init__.py
│   └── main.py
├── Dockerfile
└── requirements.txt

5. Docker 이미지 빌드

이제 도커 이미지를 빌드하고 실행합니다.

docker build -t test-image:latest .

빌드된 시간을 보면 23.9초가 걸린 것을 확인할 수 있습니다.

이후 소스코드를 변경한 후 다시 이미지를 빌드해보겠습니다.

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
def welcome_root():
    return "Hello world!"

중간에 CACHED 라는 명령어 줄을 확인해보면 캐시가 된 것을 확인할 수 있고 빌드 시간이 16.5초로 확실히 감소된 것을 확인할 수 있습니다.

만약 방금 생성한 이미지에서 컨테이너를 실행하려면 다음과 같은 명령어로 실행합니다.

docker run -p 8000:8000 test-image

reference

• Docker python
• Docker reference
• FastAPI-Docker
• https://noisrucer.github.io/posts/dockerize/

이전 포스팅에서 textual을 이용해 간단한 TUI app을 만들어보았습니다. 이번 포스팅에서는 이 라이브러리를 사용자들이 쉽게 활용할 수 있도록 배포하는 방법에 대해 알아보겠습니다.

2. 환경설정

먼저 poetry를 사용하여 PyPI에 배포하기 위한 환경을 설정합니다.

pip install pipx

pipx install poetry

이후 바탕화면에서 새로운 poetry 프로젝트를 생성합니다.

poetry new textual_app

생성한 poetry 프로젝트로 이동하여 코드를 작성할 편집기로 폴더를 열어줍니다! (예: vscode)

cd textual-app
code .

프로젝트의 구조는 다음과 같습니다.

> tests
> textual_app
pyproject.toml
README.md

3. Textual app 생성

간단한 설정이 완료되었으므로 textual app을 만들어보겠습니다. 먼저 필요한 라이브러리를 설치합니다.

poetry add "textual[dev]"

설치된 패키지는 pyproject.toml의 dependencies에 추가됩니다. pyproject.toml은 Python 프로젝트의 빌드 시스템 요구 사항을 담은 파일입니다.

[tool.poetry]
name = "textual-app"
version = "0.1.0"
description = ""
authors = ["Earthquakoo <cream5343@gmail.com>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.9"
textual = {extras = ["dev"], version = "^0.44.1"}


[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

필요한 라이브러리가 설치되었다면 poetry 가상환경을 실행합니다.

poetry shell

이제 저번 포스팅에서 만들었던 코드를 가져와 textual_app 폴더에서 main.py 파일을 생성하고 붙여넣기 해줍니다.

from textual import on
from textual.app import App, ComposeResult
from textual.screen import Screen
from textual.containers import Container, Vertical
from textual.widgets import Input, RichLog, Button


class HomeScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Container(
            Button("Get Started", id="get_started", classes="button_widget"),
            Button("Exit", id="exit", classes="button_widget"),
            classes="home_screen_container",
        )

    @on(Button.Pressed)
    def button_pressed_handler(self, event: Button.Pressed) -> None:
        if event.button.id == "get_started":
            self.app.push_screen(ChatScreen())
        elif event.button.id == "exit":
            self.app.exit()


class ChatScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Vertical(
            RichLog(classes="richlog_widget"),
            Input(placeholder="Enter chat", classes="input_widget"),
            classes="vertical_layout"
        )

    @on(Input.Submitted)
    def input_submitted_handler(self, event: Input.Submitted):
        log = self.query_one(RichLog)
        log.write(f"earthquake: {event.value}")
        input = self.query_one(Input)
        input.value = ""    



class ChatApp(App):
    CSS_PATH = "main.css"

    def on_mount(self) -> None:
        self.app.push_screen(HomeScreen())


if __name__  == "__main__":
    app=ChatApp()
    app.run()

위와 같은 예시 프로젝트를 실행하기 위해서 터미널에 아래와 같이 입력했습니다. 하지만 이것은 사용들이 사용하기엔 조금 불편함이 있습니다. 이를 수정해보도록 하겠습니다.

python textual_app/main.py

4. 패키지 관리

이제 pyproject.toml에서 몇 가지를 수정해야 합니다.

• version은 0.1.0으로 초기화 되어있습니다.
• description은 app에 대한 간단한 설명을 추가할 수 있습니다.
• [tool.poetry.scripts]로 패키지를 실행하는 명령을 정의할 수 있습니다.

여기서 중요한 것은 [tool.poetry.scripts] 입니다. 이전에 프로젝트를 실행하기 위해 터미널에서 python textual_app/main.py 로 실행했습니다. 그러나 이것은 조금 번거로우며 사용자가 길게 늘어진 패키지 타이핑을 실행하는 것은 좋지 못한 예입니다.

따라서 해당 패키지를 설치했을 때 간단한 타이핑으로 이 app을 실행할 수 있도록 scripts를 추가합니다.

[tool.poetry]
name = "textual-app"
version = "0.1.0"
description = "Simple textual app"
authors = ["Earthquakoo <cream5343@gmail.com>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.9"
textual = {extras = ["dev"], version = "^0.44.1"}

[tool.poetry.scripts]
textual-app = "textual_app.main:main"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

scripts를 설정했다면 main.py에서 해당 app의 경로를 정의한 것을 삭제하고 패키지를 실행하는 함수를 생성합니다.

from textual import on
from textual.app import App, ComposeResult
from textual.screen import Screen
from textual.containers import Container, Vertical
from textual.widgets import Input, RichLog, Button


class HomeScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Container(
            Button("Get Started", id="get_started", classes="button_widget"),
            Button("Exit", id="exit", classes="button_widget"),
            classes="home_screen_container",
        )

    @on(Button.Pressed)
    def button_pressed_handler(self, event: Button.Pressed) -> None:
        if event.button.id == "get_started":
            self.app.push_screen(ChatScreen())
        elif event.button.id == "exit":
            self.app.exit()


class ChatScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Vertical(
            RichLog(classes="richlog_widget"),
            Input(placeholder="Enter chat", classes="input_widget"),
            classes="vertical_layout"
        )

    @on(Input.Submitted)
    def input_submitted_handler(self, event: Input.Submitted):
        log = self.query_one(RichLog)
        log.write(f"earthquake: {event.value}")
        input = self.query_one(Input)
        input.value = ""    



class ChatApp(App):
    CSS_PATH = "main.css"

    def on_mount(self) -> None:
        self.app.push_screen(HomeScreen())


def main():
    app = ChatApp()
    app.run()


# if __name__ == "__main__":
#     app = ChatApp()
#     app.run()

5. 패키지 테스트 및 배포

테스트를 위해 로컬에서 패키지를 설치하고 실행해봅니다.

poetry install

패키지가 제대로 설치되었다면 아래의 textual-app 명령어로 간단히 패키지를 실행할 수 있습니다!

textual-app

이제 다른 사람들이 해당 패키지를 다운로드할 수 있도록 PyPI에 배포를 해보겠습니다.

poetry build

poetry build 명령어를 실행하면 프로젝트 구조는 아래와 같아집니다.

> dist
> tests
> textual_app
pyproject.toml
README.md

이제 PyPI사이트에서 계정을 생성하고 아래와 같이 API 토큰을 등록해주고 발급 받은 토큰은 따로 저장해둡니다.

아래의 커맨드와 함께 저장한 토큰을 입력합니다.

poetry config pypi-token.pypi <Your api token>

이제 패키지를 배포합니다.

poetry publish --build

PyPI에서 자신의 패키지명을 검색하면 배포가 된 것을 확인할 수 있습니다.

터미널에서 배포한 패키지를 설치해서 테스트합니다.

pip install textual-app

textual-app

6. 패키지 업데이트

만약 수정할 것이 있거나 새로운 기능이 생겨 업데이트가 필요하다면 pyproject.toml에서 version과 description을 수정합니다.

[tool.poetry]
name = "textual-app"
version = "0.1.1"
description = "Update textual app"
authors = ["Earthquakoo <cream5343@gmail.com>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.9"
textual = {extras = ["dev"], version = "^0.44.1"}

[tool.poetry.scripts]
textual-app = "textual_app.main:main"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

또한 textual_app/__init__.py의 __version__도 함께 수정합니다.

__version__ = "0.1.1"

다시 아래의 커맨드로 업데이트된 패키지를 배포합니다.

poetry publish --build

이후 패키지를 업데이트하려면 다음을 실행합니다.

pip install textual-app --upgrade

이전 포스팅에선 간단한 TUI app을 만들었습니다. 하지만 이는 너무 간단해보입니다. 그렇기에 조금 심화과정으로 저의 Jobdam 에서 적용되었던 몇 가지 기능을 추가해 TUI app을 꾸며보도록 하겠습니다.

2. Screen

Textual에서 Screen은 터미널의 크기를 차지하는 위젯의 컨테이너입니다. 특정 앱에는 여러 개의 화면이 있을 수 있지만 한 번에 하나의 화면만 활성화됩니다.

먼저 앱을 실행했을 때 맨 처음 보이는 화면을 만들어보겠습니다.

from textual.screen 에서 Screen 위젯 컨테이너를 가져와 특정 화면을 만들 수 있습니다.

from textual.app import App, ComposeResult
from textual.screen import Screen
from textual.widgets import Welcome

class HomeScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Welcome()

화면을 터미널 상에 출력하고 싶을 때는 push_screen이라는 메소드를 이용해 화면을 출력할 수 있습니다.

...
class ChatApp(App):

    def on_mount(self) -> None:
        self.app.push_screen(HomeScreen())


if __name__  == "__main__":
    app=ChatApp()
    app.run()

on_mount 메소드를 이용해 앱이 실행될 때 가장 먼저 마운트되는 것을 push_screen으로 설정한 뒤 앱을 실행하면 다음과 같은 결과를 얻을 수 있습니다.

2-1. HomeScreen

Textual은 빠른 스타일링을 위해 여러 컨테이너 위젯을 지원합니다.

• Horizontal: 가로 레이아웃을 위한 컨테이너
• Vertical: 세로 레이아웃을 위한 컨테이너
• Grid: 그리드 레이아웃을 위한 컨테이너
• Container: 수직 레이아웃을 위한 컨테이너

위는 대표적인 textual의 컨테이너 위젯이며 Textaul container공식문서에서 더 많은 컨테이너를 확인할 수 있습니다.

저는 홈화면에서 container 위젯으로 수직 레이아웃을 만들고 컨테이너 내에 위젯의 css 스타일을 적용하기 위해 classes와 css 파일인 main.css 만들어주었습니다.

Button의 classes 또한 위젯의 스타일을 지정하는 것이고 id는 나중에 상호작용을 위해 쓰일 예정입니다.

그리고 이 앱에 css 파일을 적용하기 위해선 CSS_PATH에 경로를 제공하면 됩니다.

from textual.app import App, ComposeResult
from textual.screen import Screen
from textual.containers import Container
from textual.widgets import Button

class HomeScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Container(
            Button("Get Started", id="get_started", classes="button_widget"),
            Button("Exit", id="exit", classes="button_widget"),
            classes="home_screen_container",
        )

class ChatApp(App):
    CSS_PATH = "main.css"

    def on_mount(self) -> None:
        self.app.push_screen(HomeScreen())

위젯의 css 스타일을 적용한 방식이 더 궁금하다면 Textual css style, Textual text css를 참고해주세요

HomeScreen {
    background: black;
}

HomeScreen .home_screen_container {
    align: center middle;
}

HomeScreen .button_widget {
    width: 10%;
    margin: 1 1 1 1;
    border: blank white;
    background: black;
}

위와 같이 설정이 끝났다면 아래와 같은 결과를 얻을 수 있습니다.

하지만 아직 버튼을 눌러도 아무런 반응을 보이지 않습니다. 이것은 후에 설정을 해주겠습니다.

2-2. ChatScreen

이제 이전에서 만들었던 간단한 채팅화면을 적용할 차례입니다.

홈화면을 만들었던 것과 비슷하게 Vertical 컨테이너를 이용해 수직 레이아웃을 만들어 그 안에 위젯들을 배치시킵니다.

class ChatScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Vertical(
            RichLog(classes="richlog_widget"),
            Input(placeholder="Enter chat", classes="input_widget"),
            classes="vertical_layout"
        )

아래는 ChatScreen의 css 구성입니다.

ChatScreen {
    background: black;
}

ChatScreen .vertical_layout {
    align: center middle;
    background: black;
}

ChatScreen .input_widget {
    width: 100%;
    margin-top: 1;
    margin-bottom: 1;
    border: white;
}

ChatScreen .richlog_widget {
    align: center middle;
    background: black;
    scrollbar-size-vertical: 1;
    scrollbar-color: white;
    border: white;
}

화면의 구성은 모두 끝이 났습니다. 하지만 아직 버튼이나 인풋의 상호작용이 일어나지 않습니다. 이제 이벤트 핸들러를 통해 상호작용을 할 수 있도록 해보겠습니다.

3. 상호작용

on 데코레이터를 이용해 이벤트 핸들러로 동작하게 합니다.

만약 event.button.id가 get_started라면 ChatScreen을 화면에 출력하고 event.button.id 가 exit이라면 앱을 종료합니다.

from textual import on

class HomeScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Container(
            Button("Get Started", id="get_started", classes="button_widget"),
            Button("Exit", id="exit", classes="button_widget"),
            classes="home_screen_container",
        )

    @on(Button.Pressed)
    def button_pressed_handler(self, event: Button.Pressed) -> None:
        if event.button.id == "get_started":
            self.app.push_screen(ChatScreen())
        elif event.button.id == "exit":
            self.app.exit()

ChatScreen도 상호작용을 위해 코드를 추가해줍니다.

class ChatScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Vertical(
            RichLog(classes="richlog_widget"),
            Input(placeholder="Enter chat", classes="input_widget"),
            classes="vertical_layout"
        )

    @on(Input.Submitted)
    def input_submitted_handler(self, event: Input.Submitted):
        log = self.query_one(RichLog)
        log.write(f"earthquake: {event.value}")
        input = self.query_one(Input)
        input.value = ""    

전체 코드는 다음과 같습니다.

from textual import on
from textual.app import App, ComposeResult
from textual.screen import Screen
from textual.containers import Container, Vertical
from textual.widgets import Input, RichLog, Button


class HomeScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Container(
            Button("Get Started", id="get_started", classes="button_widget"),
            Button("Exit", id="exit", classes="button_widget"),
            classes="home_screen_container",
        )

    @on(Button.Pressed)
    def button_pressed_handler(self, event: Button.Pressed) -> None:
        if event.button.id == "get_started":
            self.app.push_screen(ChatScreen())
        elif event.button.id == "exit":
            self.app.exit()


class ChatScreen(Screen):

    def compose(self) -> ComposeResult:
        yield Vertical(
            RichLog(classes="richlog_widget"),
            Input(placeholder="Enter chat", classes="input_widget"),
            classes="vertical_layout"
        )

    @on(Input.Submitted)
    def input_submitted_handler(self, event: Input.Submitted):
        log = self.query_one(RichLog)
        log.write(f"earthquake: {event.value}")
        input = self.query_one(Input)
        input.value = ""    



class ChatApp(App):
    CSS_PATH = "main.css"

    def on_mount(self) -> None:
        self.app.push_screen(HomeScreen())


if __name__  == "__main__":
    app=ChatApp()
    app.run()

이제 앱을 실행해 Get_started를 클릭하여 ChatScreen으로 이동하면 다음과 같은 화면으로 간단한 채팅앱이 구현된 것을 확인할 수 있습니다.

다음 포스팅에선 구축한 앱을 PyPI에 배포하는 법을 살펴보겠습니다!

코딩을 하면서 그래픽 환경인 CLI, GUI, TUI에 대해 들어보신 적이 있을 것입니다.

• CLI (Command Line Interface): 명령어를 입력하여 상호작용하는 환경으로, 윈도우에는 cmd, 리눅스에는 terminal이 대표적입니다.
• GUI (Graphic User Interface): 그래픽으로 사용자가 상호작용하는 환경으로, UI(User Interface)에 시각적인 개념이 추가된 용어입니다.
• TUI (Text User Interface): 텍스트를 통해 사용자가 상호작용하는 환경으로, 주로 터미널에서 사용되며 리눅스의 vi(vim) 편집기가 대표적입니다. (위키피디아에서는 Text User Interface를 컴퓨터 터미널의 속성에 대한 의존성을 반영하기 위해 Terminal User Interface라고도 언급하고 있습니다.)

하지만 TUI는 종종 텍스트가 주를 이루어 한 눈에 들어오지 않아 쉽게 와닿지 않을 수 있습니다. 따라서, 이번에는 Textual을 활용하여 시각적으로 화려한 TUI 앱을 구축해보려 합니다.

2. Project 생성

os: window, terminal: git bash

python -m venv venv
source venv/scripts/activate

아래의 커맨드로 개발용 textual 패키지를 설치해줍니다.

pip install "textual-dev"

만약 textual 패키지 자체를 다운로드하거나 예제를 테스트하고 싶다면 다음 명령어로 textual 패키지를 설치할 수 있습니다. (TUI 앱을 개발할 목적이라면 앞서 언급한 명령어를 사용하세요)

pip install "textual"

3. 간단한 TUI app 만들기

이제 Textual을 이용해 문구를 입력하면 화면에 출력되는 간단한 TUI app을 만들어보겠습니다.

from textual import on
from textual.app import App, ComposeResult
from textual.widgets import Input, RichLog


class ChatApp(App):

    def compose(self) -> ComposeResult:
        yield RichLog()
        yield Input()

    @on(Input.Submitted)
    def on_input_submitted(self, event: Input.Submitted):
        input = self.query_one(Input)
        log = self.query_one(RichLog)
        log.write(f"earthquake: {event.value}")
        input.value = ""


if __name__  == "__main__":
    app=ChatApp()
    app.run()

compose 메소드는 반복 가능한 인스턴스나 위젯, 혹은 위젯의 목록을 반환합니다. 여기서는 위젯을 생성하여 메서드를 generator로 만드는 것이 편리합니다.

입력을 받기 위해 Input 위젯과 입력받은 결과를 화면에 출력하기 위한 RichLog 위젯을 생성했습니다.

...
def compose(self) -> ComposeResult:
    yield RichLog()
    yield Input()

• on_input_submitted는 이벤트 핸들러로, 이벤트 핸들러는 키 누르기, 마우스 클릭 등과 같은 이벤트에 대한 응답으로 textual에서 호출하는 메소드입니다. 여기서는 Input에서 리턴값이 생겼을 때의 이벤트를 의미합니다.
  • on_input_submitted처럼 함수명을 제시해서 이벤트 핸들러로 사용할 수도 있지만 @on 이라는 데코레이터를 이용해서도 이벤트 핸들러처럼 사용할 수 있습니다. 아래의 두 가지 예시는 동일한 동작을 수행합니다.

...
@on(Input.Submitted)
def input_submitted_handler(self, event: Input.Submitted):
...

...
def on_input_submitted(self, event: Input.Submitted):
...

• query_one 메소드는 generator 또는 유형과 일치하는 단일 위젯을 가져옵니다.
• 3번째줄에선 RichLog 위젯을 가져와 RichLog의 입력 메소드인 write로 Input에서 전달받은 값을 화면에 출력합니다.
• 이후 Input의 제출이 완료되면 Input 칸의 value를 초기화시켜줍니다.

...
input = self.query_one(Input)
log = self.query_one(RichLog)
log.write(f"earthquake: {event.value}")
input.value = ""
...

실행 결과로 "hello world"를 입력하면 아래와 같은 결과를 얻을 수 있습니다.

더 자세한 정보는 Textual공식문서에서 확인이 가능합니다.