[FastAPI] DNS 측정 및 설정 API

Yoo·2025년 7월 31일

목표: 각 DNS 서버가 사용자로부터 입력받은 도메인에 접근하는 시간을 측정해서 최적의 DNS 서버로 바꿔주는 API 개발

클래스, 라이브러리 추가

from fastapi import FastAPI, HTTPException, Query

FastAPI : FastAPI 프레임워크를 생성하기 위한 클래스
HTTPException : 예외 발생 시 HTTP 에러를 응답하게 해주는 클래스
Query : 쿼리값을 받을 때 사용

import dns.resolver, time
from auto_dns import set_dns, reset_dns
from datetime import datetime

dns.resolver : dnspython 패키지에 포함된 모듈
time : 시간 관련 함수 제공해주는 라이브러리

FastAPI 생성

app = FastAPI()

dns_servers = {
    "Google": "8.8.8.8",
    "KT": "168.126.63.1",
    "SKB": "219.250.36.130",
    "LGU": "164.124.101.2",
    "KISA": "203.248.252.2"
}

app은 FastAPI 객체

직접 측정하고싶은 DNS 서버의 IP를 dns_servers에 딕셔너리 형태로 저장

measure API

기능
DNS 서버 별로 도메인에 접근하는 시간 측정

@app.get("/measure")
def measure_dns(domain: str = Query(...), count: int = Query(5, gt=0)):

@app.get("/measure")
HTTP get 요청이 /measure 경로로 오면 measure_dns 함수 실행
get은 데이터 조회할 때 사용

domain: str = Query(...)
domain은 쿼리 문자열 값이고 필수로 입력받아야 함

count : int = Query(5, gt=0)
count는 쿼리 정수형 값이고 기본값은 5, 0보다 커야 한다는 유효성 검사

localhost.8000/measure?domain=naver.com&count=3

각 DNS 서버로 도메인 응답시간 측정

records = []
    for name, ip in dns_servers.items():
        resolver = dns.resolver.Resolver()
        resolver.nameservers = [ip]

        times = []
        for i in range(count):
            try:
                start = time.time()
                resolver.resolve(domain, 'A')
                end = time.time()
                elapsed = round((end - start) * 1000, 2)
                times.append(elapsed)
            except:
                times.append(float('inf'))

        avg = round(sum(times) / len(times), 2)
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        records.append({
            "측정 시간": now,
            "DNS 서버": name,
            "도메인": domain,
            "평균 응답 시간(ms)": avg
        })

resolver = dns.resolver.Resolver()
DNS를 측정할 수 있게 객체를 만듬

resolver.nameservers = [ip]
속도를 측정할 DNS ip로 설정

resolver.resolve(domain, 'A')
A는 IPv4 주소

elapsed = round((end - start) * 1000, 2)
end 시간 - 시작 시간 * 1000을 소수 둘째자리로 변형

출력

return {
        "도메인": domain,
        "측정 횟수": count,
        "결과": records,
    }

apply API

기능
사용자가 원하는 DNS 서버로 바꿔줌

@app.get("/apply")
def dns_apply(server : str = Query(...)):
    if server not in dns_servers:
        raise HTTPException(status_code=400, detail="지원되지 않는 DNS 서버입니다.")

    server = server.strip()
    try:
        set_dns(dns_servers[server])
        status = f'{server}로 설정 완료'
    except Exception as e:
        status = f'{server}로 설정 실패'

    return {"message" : status}

@app.get("/apply")
def dns_apply(server : str = Query(...)):
HTTP get 요청이 /apply 경로로 오면 dns_apply 함수 실행
server는 쿼리 문자열 값이고 필수로 입력 받아야 함

try:
set_dns(dns_servers[server])
status = f'{server}로 설정 완료'
except Exception as e:
status = f'{server}로 설정 실패'

서버의 IP를 set_dns 함수에 전달하여 DNS 서버를 바꿈

현재 사용중인 DNS 서버

입력

http://localhost:8000/apply?server=LGU

출력

reset API

기능
사용자의 DNS 서버를 리셋

@app.get("/reset")
def reset_dns_api():
    try:
        reset_dns()
        return {
            "message" : "DNS 리셋 완료"
        }
    except Exception as e:
        return{
            "message" : "DNS 리셋 실패"
        }

입력

http://localhost:8000/reset

출력

코드

from fastapi import FastAPI, HTTPException, Query
import dns.resolver, time
from datetime import datetime
from auto_dns import set_dns, reset_dns

app = FastAPI()

dns_servers = {
    "Google": "8.8.8.8",
    "KT": "168.126.63.1",
    "SKB": "219.250.36.130",
    "LGU": "164.124.101.2",
    "KISA": "203.248.252.2"
}

@app.get("/measure")
def measure_dns(domain: str = Query(...), count: int = Query(5, gt=0)):
    records = []
    for name, ip in dns_servers.items():
        resolver = dns.resolver.Resolver()
        resolver.nameservers = [ip]

        times = []
        for i in range(count):
            try:
                start = time.time()
                resolver.resolve(domain, 'A')
                end = time.time()
                elapsed = round((end - start) * 1000, 2)
                times.append(elapsed)
            except:
                times.append(float('inf'))

        avg = round(sum(times) / len(times), 2)
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        records.append({
            "측정 시간": now,
            "DNS 서버": name,
            "도메인": domain,
            "평균 응답 시간(ms)": avg
        })   

    return {
        "도메인": domain,
        "측정 횟수": count,
        "결과": records,
    }  

@app.get("/reset")
def reset_dns_api():
    try:
        reset_dns()
        return {
            "message" : "DNS 리셋 완료"
        }
    except Exception as e:
        return{
            "message" : "DNS 리셋 실패"
        } 

@app.get("/apply")
def dns_apply(server : str = Query(...)):
    if server not in dns_servers:
        raise HTTPException(status_code=400, detail="지원되지 않는 DNS 서버입니다.")

    server = server.strip()
    try:
        set_dns(dns_servers[server])
        status = f'{server}로 설정 완료'
    except Exception as e:
        status = f'{server}로 설정 실패'

    return {"message" : status}