Python Log Class

Python에는 기본적으로 훌륭한 Log 클래스들이 존재하지만, 커스트마이즈를 쉽게 하기 위해서는 간단한 Log 클래스 하나 정도는 갖고 있는게 좋을 듯 합니다.

import threading
from datetime import datetime


class Log:
    __BLACK = "\033[30m"
    __RED = "\033[31m"
    __GREEN = "\033[32m"
    __YELLOW = "\033[33m"
    __BLUE = "\033[34m"
    __MAGENTA = "\033[35m"
    __CYAN = "\033[36m"
    __WHITE = "\033[37m"
    __UNDERLINE = "\033[4m"
    __RESET = "\033[0m"

    __lock = threading.Lock()

    @classmethod
    def d(cls, message):
        Log.__print_log(Log.__GREEN, message)

    @classmethod
    def i(cls, message):
        Log.__print_log(Log.__YELLOW, message)

    @classmethod
    def w(cls, message):
        Log.__print_log(Log.__MAGENTA, message)

    @classmethod
    def e(cls, message):
        Log.__print_log(Log.__RED, message)

    @classmethod
    def __print_log(cls, color: int, message: str):
        with Log.__lock:
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
            print(f"{color}[{timestamp}] {message}{Log.__RESET}")

Python에서 Celery 사용하기

Celery는 Task Queue입니다. RabbitMQ나 Redis 등을 백엔드로 사용하며 그 위에서 Wrapper로 추상화된 인터페이스를 제공해줍니다.

설치

pip3 install redis
pip3 install celery

tasks.py

Task 실행 요청을 수신하는 서버측 코드입니다.

import time

from celery import Celery

BROKER_URL = 'redis://localhost:6379/0'
BACKEND_URL = 'redis://localhost:6379/1'
app = Celery('tasks', broker=BROKER_URL, backend=BACKEND_URL)


@app.task(name="snowdeer_add")
def add(x, y):
    for i in range(1, 10):
        print("Calculating ...")
        time.sleep(0.5)

    return x + y

실행은 일반적인 python3로 하는 것이 아니라 아래의 명령어로 실행합니다.

$ celery -A tasks worker --loglevel=info

 -------------- celery@Choongs-MacBook-Pro.local v5.2.1 (dawn-chorus)
--- ***** ----- 
-- ******* ---- macOS-12.0.1-arm64-arm-64bit 2021-12-25 19:59:29
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x10273e430
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     redis://localhost:6379/1
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . snowdeer_add

[2021-12-25 19:59:29,936: INFO/MainProcess] Connected to redis://localhost:6379/0
[2021-12-25 19:59:29,945: INFO/MainProcess] mingle: searching for neighbors
[2021-12-25 19:59:30,982: INFO/MainProcess] mingle: all alone
[2021-12-25 19:59:31,035: INFO/MainProcess] celery@snowdeer-MacBook-Pro.local ready.

client.py

from tasks import add

f = add.apply_async(args=[2, 3, ])
result = f.get(timeout=10)

print(f"result: {result}")

단순히 Task만 요청하고 싶을 때는 add.apply_async(args=[2, 3, ]) 대신 add.delay(2, 3)과 같은 코드로 대신할 수도 있습니다.

M1에 Docker 설치하는 방법

Docker 다운로드

Docker가 M1에서 동작할 수 있도록 정식 버전이 릴리즈되어 있습니다.

여기에서 다운로드할 수 있습니다.

Docker Subscription Service Agreement

Accept를 누르고 계속 진행합니다.

Tutorial

설치를 하고나면 간단한 튜토리얼을 해볼 수 있습니다.

clone, build, run, share에 대한 간단한 명령어를 확인할 수 있습니다.

터미널에서 확인

터미널에서도 Docker 명령어가 잘 동작하는 것을 확인합니다.

실제 동작 확인

여기서는 RabbitMQ를 이용해서 테스트 해보도록 하겠습니다. 아래 명령어를 입력하면 RabbitMQ 이미지까지 자동으로 다운로드 후 실행까지 합니다.

$ docker run -d --hostname rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management

Unable to find image 'rabbitmq:3-management' locally
3-management: Pulling from library/rabbitmq
a39c84e173f0: Pull complete
7d3994c28245: Pull complete
10c911d5c079: Pull complete
769f1e4dc40b: Pull complete
2090f091d001: Pull complete
f9e692861b3e: Pull complete
d26c4b0e32ac: Pull complete
3cf30fecd6f0: Pull complete
c5b6ca5b444e: Pull complete
aa1dff4734e4: Pull complete
Digest: sha256:4c4b66ad5ec40b2c27943b9804d307bf31c17c8537cd0cd107236200a9cd2814
Status: Downloaded newer image for rabbitmq:3-management
501592f6fdc53e1da8ff527b2b4cf7853ce28bf05dd06e3c62fbac6e747e4945

잘 동작하는지 확인해봅니다.

$ docker ps -a

CONTAINER ID   IMAGE                   COMMAND                  CREATED          STATUS          PORTS                                                                                                         NAMES
501592f6fdc5   rabbitmq:3-management   "docker-entrypoint.s…"   28 seconds ago   Up 27 seconds   4369/tcp, 5671/tcp, 0.0.0.0:5672->5672/tcp, 15671/tcp, 15691-15692/tcp, 25672/tcp, 0.0.0.0:15672->15672/tcp   rabbit

그리고 사파리나 크롬 등에서 http://localhost:15672/에 접속해서 다음 화면이 잘 나오는지 확인합니다.

Redis Pub/Sub 예제

Redis 설치

Redis는 Docker를 이용해서 설치해줍니다.

docker run --name redis -d -p 6379:6379 redis

subscriber.py

import redis

r = redis.Redis(host="localhost", port=6379, db=0)
s = r.pubsub()

s.subscribe("snowdeer_channel")

while True:
    print("waiting message...")
    res = s.get_message(timeout=5)
    if res is not None:
        print(f"res: {res}")

publisher.py

import redis
import datetime

r = redis.Redis(host="localhost", port=6379, db=0)

msg = f"[{datetime.datetime.now()}] hello, snowdeer +___+"
r.publish(channel="snowdeer_channel",
          message=msg)

print(f"Message is sent !!\n{msg}")

여러 개의 Topic을 Subscription 하는 예제

import redis


def main():
    r = redis.Redis(host="localhost", port=6379, db=0)
    s = r.pubsub()

    s.subscribe('chat')
    s.subscribe('event')

    while True:
        print('waiting message...')
        res = s.get_message(timeout=5)
        if res is not None:
            if res['type'] == 'message':
                handle_message(res['channel'], res['data'])


def handle_message(topic: str, message: str):
    if topic == 'chat':
        handle_message_for_chat(message)
    elif topic == 'event':
        handle_message_for_event(message)


def handle_message_for_chat(message: str):
    print(f'chat message: {message}')


def handle_message_for_event(message: str):
    print(f'event message: {message}')


if __name__ == '__main__':
    main()

터미널에서 code 명령어 실행할 수 있도록 설정

alias 등을 이용해서 터미널에서 vscode를 실행할 수도 있지만, 아예 vscode 프로그램 내에서 설정하는 방법이 더 깔끔한 것 같습니다.

vscode를 실행한다음 Command + Shift + P를 입력해서 Command Palette를 실행한 다음 shell이라고 입력하면, Shell Command: Install 'code' in PATH를 선택합니다. 그러면 이제 터미널에서 code 명령어를 이용해서 vscode를 실행할 수 있습니다.