[Program][보족] 현장 PC에서 서버를 켜면 자동으로 감시가 시작되게 만든 방법

이 글이 나한테 해당되는지 먼저 확인해보세요

이런 상황이 하나라도 해당된다면, 이 글이 도움이 될 겁니다.

• FastAPI 서버가 켜진 직후부터 자동으로 뭔가 실행되게 만들고 싶다
• API 요청이 들어올 때만 동작하는 게 아니라, 서버가 알아서 주기적으로 작업하게 하고 싶다
• 백그라운드 루프를 만들었는데 에러가 한 번 나면 루프가 멈춰버린다
• JavaScript의 setInterval처럼 동작하는 걸 Python 서버에서 구현하고 싶다
• 서버를 켜면 자동으로 PLC나 외부 장비 감시가 시작되게 하고 싶다

PLCLink에서 서버가 켜지는 순간부터 PLC를 1초마다 감시하게 만든 방법을 코드와 함께 정리했습니다.

---

용어 먼저 짚고 넘어갈게요

asyncio란

Python에서 여러 작업을 동시에 처리할 수 있게 해주는 내장 라이브러리입니다.

정확히는 "동시에" 처리하는 게 아니라 "번갈아가며 빠르게" 처리하는 방식인데, 사용자 입장에서는 동시에 도는 것처럼 보입니다.

식당 비유로 설명하면, 요리사가 한 명인데 국을 끓이는 동안 기다리지 않고 그 시간에 채소를 썰다가, 국이 끓으면 다시 국으로 돌아오는 방식입니다.

실제로 두 일을 동시에 하는 건 아니지만 기다리는 시간을 낭비하지 않아요.

PLC 통신처럼 "보내고 응답 기다리는" 작업에 잘 맞습니다.

기다리는 동안 다른 API 요청을 처리할 수 있으니까요.

 

lifespan이란

FastAPI 서버가 시작될 때와 종료될 때 각각 한 번씩 실행할 코드를 등록하는 방법입니다.

식당으로 치면 "문 열기 전에 할 일"과 "문 닫은 후에 할 일"을 정해두는 겁니다.

서버가 시작하자마자 PLC 감시를 시작하게 하려면 lifespan에 등록합니다.

 

create_task란

asyncio에서 어떤 함수를 백그라운드에서 독립적으로 실행시키는 방법입니다.

함수를 호출해서 결과를 기다리는 게 아니라, "이건 알아서 계속 돌아" 하고 넘기는 방식입니다.

한 번 시작하면 서버가 종료될 때까지 계속 실행됩니다.

---

문제의 시작 : API는 요청이 올 때만 동작한다

FastAPI를 처음 배울 때 자연스럽게 만드는 구조는 이렇습니다.

@app.get("/plc/value")
async def get_plc_value():
    value = await read_plc()
    return {"value": value}

누군가 /plc/value를 호출하면 PLC에서 값을 읽어서 돌려줍니다.

아무도 호출 안 하면 아무 일도 안 일어납니다.

이 구조로는 "아무도 화면을 보고 있지 않아도 1초마다 PLC를 읽어서 알람 조건을 판단하고 이력을 쌓는" 게 불가능합니다.

요청이 있어야만 동작하니까요.

필요한 건 요청과 무관하게 서버 혼자 계속 도는 루프입니다.

---

가장 단순한 접근 : while True 루프

개념 자체는 단순합니다. 무한히 반복하면서 PLC를 읽고, 1초 기다리고, 다시 읽고.

async def alarm_poller_loop():
    while True:
        value = await read_plc()
        if value > threshold:
            save_alarm(value)
        await asyncio.sleep(1)

asyncio.sleep(1)이 핵심입니다. 일반 time.sleep(1)을 쓰면 서버 전체가 1초 동안 멈춥니다.

그 1초 동안 들어오는 API 요청을 전혀 처리 못해요.

asyncio.sleep은 "나는 1초 쉴 테니 그동안 다른 거 처리해"라는 뜻입니다.

루프가 쉬는 동안 다른 API 요청이 들어오면 그걸 처리합니다.

이 루프를 어떻게 시작할지가 다음 문제입니다.

---

서버 시작할 때 루프를 켜는 방법 : lifespan

FastAPI에서 서버 시작 시 실행할 코드를 등록하는 방법이 lifespan입니다.

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 서버 시작 : 여기서 백그라운드 태스크 등록
    task = asyncio.create_task(alarm_poller_loop())

    yield  # 서버가 실행되는 동안 여기서 대기

    # 서버 종료 : 여기서 정리 작업
    task.cancel()

app = FastAPI(lifespan=lifespan)

yield 위쪽이 서버 시작 시 실행할 코드, 아래쪽이 서버 종료 시 실행할 코드입니다.

asyncio.create_task()가 핵심입니다.

이걸 호출하는 순간 alarm_poller_loop가 백그라운드에서 독립적으로 실행되기 시작합니다.

결과를 기다리거나 멈추지 않고 바로 다음 줄로 넘어갑니다.

루프는 백그라운드에서 계속 돌고, 서버는 평소대로 API 요청을 처리합니다.

---

루프가 에러로 죽으면 안 된다

여기서 현장에서 실제로 겪은 문제가 있습니다.

PLC 케이블이 빠졌을 때 read_plc()에서 예외(Exception)가 발생합니다.

이 예외를 잡아주지 않으면 루프 자체가 종료됩니다.

서버는 살아있지만 감시는 멈춰버리는 상태가 됩니다.

# 위험한 코드 : 예외 발생 시 루프 종료
async def alarm_poller_loop():
    while True:
        value = await read_plc()  # 여기서 예외 발생하면 루프 끝
        check_alarm(value)
        await asyncio.sleep(1)

# 안전한 코드 : 예외가 나도 루프 유지
async def alarm_poller_loop():
    while True:
        try:
            value = await read_plc()
            check_alarm(value)
        except Exception as e:
            # 에러 로그만 남기고 다음 루프로 진행
            print(f"[AlarmPoller] 오류 발생: {e}")
        await asyncio.sleep(1)  # try 바깥 : 에러 나도 1초 후 재시도

asyncio.sleep(1)을 try 블록 바깥에 둔 것도 중요합니다.

에러가 나도 1초 기다린 뒤 다시 시도합니다.

에러가 연속으로 날 때 무한정 재시도하면서 CPU를 100% 쓰는 상황을 막아줍니다.

 

케이블이 다시 꽂히면 다음 루프에서 자연스럽게 정상으로 돌아옵니다.

수동으로 재시작할 필요가 없어요.

---

여러 루프를 동시에 돌리기

PLCLink에서는 알람 감시(AlarmPoller)와 트리거 스냅샷(TriggerPoller) 두 가지를 동시에 돌립니다.

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 두 루프를 동시에 시작
    alarm_task   = asyncio.create_task(alarm_poller_loop())
    trigger_task = asyncio.create_task(trigger_poller_loop())

    yield

    # 서버 종료 시 둘 다 정리
    alarm_task.cancel()
    trigger_task.cancel()

app = FastAPI(lifespan=lifespan)

create_task를 여러 번 호출하면 각각 독립적으로 백그라운드에서 실행됩니다.

AlarmPoller는 1초 간격, TriggerPoller는 100ms 간격으로 다르게 설정할 수 있어요.

서로 영향을 주지 않습니다.

---

루프 안에서 DB 설정을 동적으로 읽기

한 가지 더 신경 쓴 부분이 있습니다.

알람 룰은 사용자가 브라우저에서 언제든지 추가하거나 삭제할 수 있습니다.

루프가 시작할 때 룰을 한 번 읽어두고 고정해두면, 룰을 바꿔도 반영이 안 됩니다.

async def alarm_poller_loop():
    while True:
        try:
            # 매 루프마다 DB에서 최신 룰을 읽어옴
            rules = get_alarm_rules_from_db()

            for rule in rules:
                value = await read_plc(rule['address'])
                if check_condition(rule, value):
                    insert_alarm_history(rule, value)

        except Exception as e:
            print(f"[AlarmPoller] 오류: {e}")

        await asyncio.sleep(1)

매 루프마다 DB에서 최신 알람 룰을 가져옵니다.

사용자가 룰을 추가하면 다음 루프(최대 1초 후)부터 바로 적용됩니다.

서버를 재시작하지 않아도 됩니다.

SQLite 읽기는 매우 빠르기 때문에 1초 루프에서 매번 읽어도 부하가 없습니다.

---

전체 흐름 정리

지금까지 설명한 구조를 한 번에 정리하면 이렇습니다.

# plclink/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

# ── 알람 감시 루프 ──────────────────────────
async def alarm_poller_loop():
    while True:
        try:
            rules = get_alarm_rules_from_db()          # 최신 룰 읽기
            for rule in rules:
                value = await read_plc(rule['address']) # PLC 조회
                if check_condition(rule, value):
                    insert_alarm_history(rule, value)   # 이력 기록
        except Exception as e:
            print(f"[AlarmPoller] {e}")
        await asyncio.sleep(1)                          # 1초 대기

# ── 서버 시작 / 종료 ────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(alarm_poller_loop())    # 백그라운드 시작
    yield                                               # 서버 실행 중
    task.cancel()                                       # 서버 종료 시 정리

app = FastAPI(lifespan=lifespan)

# ── 일반 API ────────────────────────────────
@app.get("/alarm/history")
async def get_alarm_history():
    return fetch_alarm_history_from_db()               # 브라우저 조회용

서버가 시작되면 alarm_poller_loop가 백그라운드에서 실행됩니다.

브라우저가 연결됐든 없든, 탭이 열렸든 닫혔든 루프는 계속 돕니다.

브라우저는 /alarm/history API로 이미 쌓인 이력을 조회할 뿐입니다.

---

자주 하는 실수 두 가지

첫 번째 : asyncio.sleep 대신 time.sleep 사용

import time

# 잘못된 방법 : 서버 전체가 1초 동안 멈춤
async def alarm_poller_loop():
    while True:
        read_plc()
        time.sleep(1)  # 이러면 안 됨

# 올바른 방법
async def alarm_poller_loop():
    while True:
        await read_plc()
        await asyncio.sleep(1)  # 이래야 다른 요청 처리 가능

두 번째 : lifespan 없이 startup 이벤트 사용

예전 FastAPI 버전 예제에서 @app.on_event("startup")을 사용한 코드를 많이 볼 수 있습니다.

지금도 동작하지만 deprecated(더 이상 권장하지 않음) 상태입니다.

새로 만드는 코드는 lifespan 방식을 쓰는 게 좋습니다.

# 구버전 방식 : 지금도 동작하지만 권장 안 함
@app.on_event("startup")
async def startup():
    asyncio.create_task(alarm_poller_loop())

# 현재 권장 방식
@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(alarm_poller_loop())
    yield
    task.cancel()

---

마치며

"API 서버는 요청을 받아서 응답하는 것"이라고만 생각했는데, 서버 자체가 능동적으로 일하게 만들 수 있다는 게 이 구조에서 배운 것입니다.

구현해보면 코드 자체는 길지 않습니다.

핵심은 세 줄입니다.

asyncio.create_task()    : 루프를 백그라운드에서 독립 실행
await asyncio.sleep()   : 다른 요청 처리를 막지 않는 대기
try-except              : 에러가 나도 루프가 죽지 않게

이 세 가지를 제대로 쓰면, 서버가 켜지는 순간부터 꺼지는 순간까지 PLC를 혼자 알아서 감시합니다.