fastapi에서 일정 주기로 뭔가를 실행시키는 방법이 없을까요? 예를 들어, 자정 12시마다 방문자 수를 초기화 한다던지, 통계 데이터를 수집해야 하는 경우가 있습니다. 이를 cron으로 돌려서 처리할 수도 있습니다만, 여간 귀찮은 것이 아닙니다. celery에는 worker 뿐만이 아니라, 스케쥴러를 등록해서 사용할 수도 있습니다. 간단하게 알아보겠습니다.
프로젝트 구조는 위와 같습니다. core의 celery 안에 있는 base.py로 들어가 보겠습니다.
코드를 보겠습니다. broker와 backend는 redis를 사용하였습니다. 고로 redis 주소를 적어주었습니다. 다음에, include에는 실행할 테스크가 있는 "app.sche.sche"를 적어주었습니다. 위 코드에서는 메세지를 전송하는 주체와 수신하는 주체를 중개해 주는 역할을 redis가 하게 됩니다.
다음에, celery_task의 conf의 beat_schedule에는 실행할 task에 대한 정보를 적어줍니다. app.sche.sche.sche_every_minute라는 테스크를 실행하는데요. "schedule"에 crontab()을 넣어줍니다. crontab에 아무 인자도 들어가지 않은 경우, 매 분마다 실행되게 됩니다.
매 분마다 실행시킬 task입니다. task는 "doai station"을 출력하는 것입니다. 여기까지 내용을 정리해 봅시다.
스케쥴러를 정의했는데요. 여기에 매 분마다 실행되는 sche_every_minute task를 등록해 줍니다. 이 task는 1분마다 어딘가로 전송이 될 겁니다
이제, worker에게 해당 task를 전송합니다.
매 분마다 요런 흐름으로 로직이 구성됩니다. 간단하게 테스트 해 보겠습니다.
이제 간단하게 명령어를 입력해서 테스트 해 보겠습니다. app.core.celery.base.celery_task에 실행시킬 task들이 있습니다. 뒤에 worker -B가 붙어 있습니다. 이는 embedded beat를 사용한다는 의미입니다. 간단하게 테스트 하기 위해 embed beat를 이용합시다.
1분마다 doai station이 출력되는 것을 볼 수 있어요. 다음 시간에 저 명령어에 대해서 조금 더 자세히 알아보도록 하겠습니다.
'웹 > FASTAPI' 카테고리의 다른 글
| sqlalchemy like 연산의 %와 _를 escaping 처리해 주는 startswith, endswith, contains를 알아봅시다. (0) | 2023.06.23 |
|---|---|
| fastapi sub application에 대해 알아보고 app마다 다른 middleware를 적용해 봅시다. (0) | 2023.06.13 |
| fastapi pydantic 필드 단위로 동작하는 validator를 간단하게 알아봅시다 (0) | 2023.06.09 |
| fastapi pydantic root validator를 알아봅시다. (1) | 2023.06.06 |
| fastapi alembic revision 명령어를 쓸 때 마이그레이션 버전을 커스텀하게 생성해 봅시다. (0) | 2023.05.31 |
최근댓글