[Monitoring System] 6. Fast API 로 이틀만에 백엔드 구축하기(feat. SQLAlchemy, PostgreSQL) (2) - MQTT 데이터 적재 / 로그인 기능 구현

2023.06.13 - [Project/Monitoring System] - [Monitoring System] 5. Fast API 로 이틀만에 백엔드 구축하기(feat. SQLAlchemy, PostgreSQL) (1)

[Monitoring System] 5. Fast API 로 이틀만에 백엔드 구축하기(feat. SQLAlchemy, PostgreSQL) (1)

전체 코드

https://github.com/melli0505/monitoring_system

0. 변경사항

1편을 올리고 무려 2달의 시간이 지났다. 때문에 내 모니터링 시스템에도 많은 변화가 있었는데... 사실 그렇게 많은 변화는 아니다(?). 1편에서는 테이블이 user, data, sensor 이렇게 세 가지로 구성해놨었는데, 지금은 더미데이터가 아닌 실제 데이터를 기반으로 해서 모니터링 시스템을 구축해뒀기 때문에 sensor 없이 data 의 타입에 따라 user 포함 총 7개의 데이터 테이블이 생성되었다.

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String, index=True)
    disable = Column(Boolean, index=True, default=True)

class Voltage(Base):
    __tablename__ = "voltages"

    id = Column(Integer, primary_key=True, index = True)
    time = Column(TIMESTAMP, index=True)
    voltage = Column(Float, index=True)

class Current(Base):
    __tablename__ = "currents"

    id = Column(Integer, primary_key=True, index = True)
    time = Column(TIMESTAMP, index=True)
    current = Column(Float, index=True)

class Power(Base):
    __tablename__ = "powers"

    id = Column(Integer, primary_key=True, index = True)
    time = Column(TIMESTAMP, index=True)
    power = Column(Float, index=True)

class Energy(Base):
    __tablename__ = "energies"

    id = Column(Integer, primary_key=True, index = True)
    time = Column(TIMESTAMP, index=True)
    energy = Column(Float, index=True)

class PF(Base):
    __tablename__ = "pfs"

    id = Column(Integer, primary_key=True, index = True)
    time = Column(TIMESTAMP, index=True)
    pf = Column(Float, index=True)

class Frequency(Base):
    __tablename__ = "frequencies"

    id = Column(Integer, primary_key=True, index = True)
    time = Column(TIMESTAMP, index=True)
    frequency = Column(Float, index=True)

데이터의 출처는 이 녀석이다. 이 녀석과 ESP32 계열 아두이노를 연결해서 MQTT 방식으로 데이터를 서버로 보내도록 해두었다.

pzem 004t - 중간광고 아닙니다

pzem-004t 제어를 위한 아두이노 코드는 여기에 구현해두었다. 당장 여기서 중요한 것은 아니니 넘어가도록 하자.

https://github.com/melli0505/ESP32_PZEM-004t

GitHub - melli0505/ESP32_PZEM-004t

 

1. MQTT 데이터 적재

우선 MQTT로 받아온 데이터들을 각 데이터 형식에 맞게 호환을 시켜주고, 적절하게 업데이트를 시켜주어야한다. 거두절미하고 코드를 보자.

import paho.mqtt.client as mqtt
import time
import json

from db import models
from db.base import SessionLocal

broker_address = "YOUR IP ADDRESS"

# subscriber callback

def on_message(client, userdata, message):
    db = SessionLocal()
    converted = str(message.payload.decode("utf-8"))
    value = json.loads(converted)
    print(message.topic, value['voltage'], type(value['voltage']))
    timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
    data = [models.Voltage(time=timestamp, frequency=value["voltage"]),
            models.Energy(time=timestamp, frequency=value["energy"]),
            models.Power(time=timestamp, frequency=value["power"]),
            models.Current(time=timestamp, frequency=value["current"]),
            models.Frequency(time=timestamp, frequency=value["frequency"]),
            models.PF(time=timestamp, frequency=value["pf"])]
    for db_data in data:
        db.add(db_data)
        db.commit()
        db.refresh(db_data)
    db.close()


if __name__ == "__main__":
    client = mqtt.Client("client")

    client.connect(broker_address, 1883)
    print("client connected")

    client.on_message = on_message
    client.subscribe("data")

    print("client loop started")
    client.loop_forever()

models 에서 미리 정의해둔 데이터 모델에 맞춰서 들어온 데이터들을 가공해주었다. 나 같은 경우에는 총 6개 데이터를 보내는 쪽에서 Json string 형태로 가공해서 받아왔기 때문에 처리를 이렇게 해주었는데, 각자 데이터 타입에 맞게 해주면 된다.

db.add(db_data) 와 db.refresh() 는 데이터베이스에 데이터를 올릴 때만 사용한다. 삭제의 경우에는 delete -> commit 순으로 사용한다. 잘 올라갔는지 확인해보자.

굿! 잘 들어온 것을 확인할 수 있었다. 총 6개의 데이터 테이블 모두 잘 적재되고 있다. 그렇다면 다음으로 해줄 것은, 그렇다. 서비스 구현이다.

 

2. 로그인 기능 구현

로그인 기능을 처음 구현할 때 많이 헤맸다. 어떻게 현재 접속한 사람이 로그인한 사람인지, 누구인지를 알 수 있지? 어떻게 해야 비밀번호를 암호화할 수 있지? 이런 질문들 끝에 가장 간단한 방법 하나를 적용하기로 했다. 실제 서비스에서는 이런 식으로 비밀번호를 관리하면 아주 커다란 보안 문제가 작용할 수 있으므로 주의(...) 

우선 현재 접속한 사람의 권한은 쿠키에 저장된다. 쿠키에 access token이라는 이름으로 access token을 발행해주면, 이걸 가지고 페이지를 옮겨다니면서 만료 전까지 내가 누구예요! 하고 서버에 알려줄 수 있다. 

그리고 비밀번호는 암호화해서 database에 저장되어야하고, 로그인할 때 암호화 방식에 따라 입력된 비밀번호와 암호화된 비밀번호가 매칭되는지를 확인하고, 맞다면 토큰을 발행하면 된다. 하나씩 자세히 보자. 

 

A. Hashed password 생성하기

우선 비밀번호 암호화부터 해보자. passlib의 context에서 CryptContext라는 모듈을 활용해서 생성할 것이다. 

from passlib.context import CryptContext

password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

bcrypt라는 알고리즘을 활용해서 비밀번호 암호화를 해줄 것인데, 구현이 쉬운 것에 비해서는 괜찮은 암호화 성능을 자랑한다고 한다. password_context를 설정했으면, 처음 가입해줄 때 password를 이쪽으로 보내서 암호화한 다음에 database에 들어가도록 해주어야한다. 유저 생성 함수와 함께 보자.

# create user
def create_user(db: Session, user: schemas.UserCreate):
    password = utils.get_hashed_password(user.password)
    db_user = models.User(username=user.username, hashed_password=password, email=user.email)
    # database session에 db_user 정보 추가
    db.add(db_user)
    # db에 반영
    db.commit()
    # 최신 db 정보 
    db.refresh(db_user)
    return db_user

get_hashed_password 함수는 아래와 같다.

def get_hashed_password(password: str) -> str:
    return password_context.hash(password)

짜잔. 이게 끝이다. 그렇다면 나중에 로그인을 할 때, 어떻게 맞는 비밀번호인지를 확인할까? 확인할 수 있는 함수도 제공한다. 로그인 함수를 보자.

@app.post("/login")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: Session = Depends(get_db)):
    user = user_crud.get_user_by_username(db=db, username=form_data.username)
    if user is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect email or password")
    hashed_pass = user.hashed_password
    if not utils.verify_password(form_data.password, hashed_pass):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect email or password")
    
    return {
        "access_token": utils.create_access_token(user.username),
        "refresh_token": utils.create_refresh_token(user.username)
    }

먼저 입력받은 form data에서 중복 불가능한 username을 기반으로 DB에서 유저 데이터를 찾는다. 그리고 그 유저의 hashed password를 얻어서, verify_password 함수를 통해 현재 입력된 비밀번호와 같은 값인지를 확인하도록 되어있다.

form data의 annotation이 OAuth2PasswordRequestForm으로 되어있는데, 비밀번호가 포함된... 뭐 그런 보안 정보를 form으로 주고 받을 때 fastapi.security에서 제공되는 form 형식인 것으로 기억한다. 아니라면 알려주십시오... 

def verify_password(password:str, hashed_pass:str) -> bool:
    return password_context.verify(password, hashed_pass)

verify 함수도 passlib에서 제공해주므로 사용만 하면 된다. 

 

B. Access Token 부여하기

Access token이 무엇이냐! 내가 누구인지를 서버에 알려줄 수 있는 일종의 신분증이다. 로그인한 사용자만 이용가능한 서비스를 구축하고 싶다면 로그인을 했는지, 안 했는지, 이 페이지를 볼 권한이 있는지, 없는지를 체크해줄 수 있다. 

Token 생성에는 HS256 알고리즘을 써줄 것이다. 이 때는 jose의 jwt 라이브러리를 사용할 것인데, 미리 선언해줄 것이 4가지 있다.

from jose import jwt

ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7
ALGORITHM = "HS256"
SECRET_KEY = "SECRET_KEY"

Access token 과 refresh token expire minutes는 말 그대로 만료 기한이고, 대체로  access는 30분, refresh는 일주일이라고 해서 일주일로 잡아뒀다. 알고리즘은 암호화 알고리즘을 말하는 것이고, Secret Key는 암호화할 때 사용할 고유 키인데, 외부로 공개하지 않아야한다.(나는 당장은 상관 없어서 github에는 올려놨다)

이제 access token 생성, refresh token 생성, 그리고 access token을 통해서 현재 로그인한 유저가 누구인지 알아내는 함수까지 작성해볼 차례다. 먼저 token 생성부터 보자.

def create_access_token(subject: Union[str, Any], expires_delta: int = None) -> str:
    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)

    to_encode = {"exp": expires_delta, 'sub': str(subject)}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, ALGORITHM)
    return encoded_jwt

def create_refresh_token(subject: Union[str, Any], expires_delta: int = None) -> str:
    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)

    to_encode = {"exp": expires_delta, 'sub': str(subject)}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, ALGORITHM)
    return encoded_jwt

expires delta는 외부에서 받아오지 않도록 했으므로 else문 내부의 정의로 설정된다. 그리고 암호화 인코딩할 데이터를 expires_delta와 전달받은 subject - 여기에서는 username을 json string으로 만든 뒤 토큰화해서 리턴한다.

 

C. current user 확인하기 

다음은 현재 접속 중인 유저 확인이다. 이 부분은 access token을 디코딩하기만 하면 되니 바로 보자.

async def get_current_user(token: str = Depends(get_token), db: Session = Depends(get_db)) -> models.User:
    try: 
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        token_data = user_schema.TokenPayload(**payload)

        if datetime.fromtimestamp(token_data.exp) < datetime.now():
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token Expired")
    except(jwt.JWTError, ValidationError):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate credentials")
    
    user = user_crud.get_user_by_username(db=db, username=token_data.sub)

    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Could not find user")
    
    return user

예외처리 부분을 제하고 본다면 간단하다. jwt의 decode를 활용해서 algorithm과 secret key를 넘겨주고 데이터를 받아온다. 그리고 TokenPayload라는 클래스에 데이터를 매핑해준다. TokenPayload는 아래와 같이 선언해주었다.

class TokenPayload(BaseModel):
    sub: str = None
    exp: int = None

이 토큰 데이터는 토큰을 생성할 때 암호화 인코딩할 데이터의 key값에 맞춰서 클래스를 생성해주어야한다. 그리고 이걸 기반으로 username으로 데이터를 찾아내고, 유저가 있는지, 없는지, 있다면 그 유저의 데이터를 리턴해준다.

get_token 함수로 token을 받아오도록 해뒀는데, 정의는 아래와 같다. authorization이라는 쿠키 데이터를 받아오면 그것의 6번째 인덱스부터가 token이었다.(꽤나 노가다를 했더랬다.)

def get_token(authorization: str = Header(default=None)):
    print(authorization)
    return authorization[6:]

이제 기능은 구현됐다!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! UI에 입혀보도록 하자.

 

D. 프론트엔드 만들기 

안타깝게도 UI 디자인에는 재능이 없다. 그냥 오픈된 부트스트랩 템플릿을 하나 주워올 것이다. 

https://colorlib.com/wp/template/login-form-15/

이 녀석으로 골랐다. 로그인과 회원가입은 사실상 거의 같은 형식을 갖고 있기 때문에, Form data를 어떻게 잘 api 쪽으로 넘겨주고 확인을 받아오면 된다. 먼저 회원가입이다.

1. 회원가입

html 부분부터 잠깐 보자.

<form action="#" class="signup-form">
  <div class="form-group mt-3">
    <input
      id="username"
      type="text"
      class="form-control"
      required
    />
    <label class="form-control-placeholder" for="username">
      Username
    </label>
  </div>
  <div class="form-group">
    <input
      id="email"
      type="email"
      class="form-control"
      required
    />
    <label class="form-control-placeholder" for="email">
      email
    </label>
  </div>
  <div class="form-group">
    <input
      id="password"
      type="password"
      class="form-control"
      required
    />
    <label class="form-control-placeholder" for="password">
      Password
    </label>
    <span
      toggle="#password-field"
      class="fa fa-fw fa-eye field-icon toggle-password"
    ></span>
  </div>
  <div class="form-group">
    <button
      type="submit"
      class="form-control btn btn-primary rounded submit px-3"
    >
      Sign Up
    </button>
  </div>
</form>

form 태그로 둘러싸여있는 데이터들을 id로 구분을 해줬다. 대부분 아시겠지만 input 태그 내부 속성 중 type을 지정하면 해당하는 타입을 지정할 수 있는데, email에 email 타입을 넣으면 ~~~~@~~~~.~~ 형식을 안 지키면 혼내준다. 

이제 jQuery ajax 파트로 넘어가자. 훗날에 비하면 간단하기 그지없다.

<script type="text/javascript">
  $(document).ready(function ($) {
    $(".signup-form").submit(function (event) {
      event.preventDefault();

      var body = new FormData();
      body.append("username", $("#username").val());
      body.append("password", $("#password").val());
      body.append("email", $("#email").val());

      console.log(body);

      var data = new URLSearchParams(body);

      $.ajax({
        url: "http://localhost:8000/api/users/signup",
        method: "POST",
        data: data,
        contentType: false,
        processData: false,
        headers: {
          "Content-Type": "application/x-www-form-urlencoded",
        },
        success: function (response) {
          alert("Account Created");
          location.href = "http://localhost:8000/";
        },
        error: function (response) {
          alert(response["detail"]);
        },
      });

      event.preventDefault();
    });
  });
</script>

꽤 기니까 부분부분 뜯어보겠다.

$(".signup-form").submit(function (event) {

이 form 전체가 제출이 되면 요 함수를 부르겠다는 뜻이다. 제출 버튼은 아래에 따로 만들어줬었다.

var body = new FormData();
body.append("username", $("#username").val());
body.append("password", $("#password").val());
body.append("email", $("#email").val());

var data = new URLSearchParams(body);

key-value로 매칭을 시켜줬다. 이렇게 해야 서버 쪽에서 아하 이게 무슨 데이터구나~ 하고 알아볼 수 있다. $("#id").val() 혹은 $(".class").val() 라고 써서 데이터를 가져올 수 있다. 찾아보니 URLSearchParams라는 객체로 만들어줘야 한다고 해서 묶어주었다.

$.ajax({
    url: "http://localhost:8000/api/users/signup",
    method: "POST",
    data: data,
    contentType: false,
    processData: false,
    headers: {
      "Content-Type": "application/x-www-form-urlencoded",
    },
    success: function (response) {
      alert("Account Created");
      location.href = "http://localhost:8000/";
    },
    error: function (response) {
      alert(response["detail"]);
    },
  });

이제 ajax로 이 데이터를 갖다가 api 라우터 어딘가로 날려준다. 바로 signup 으로! post 방식까지 지정해서 보내주면 

@router.post("/signup") #, response_model=schemas.User)
def create_user(username: Annotated[str, Form()], email: Annotated[str, Form()], password: Annotated[str, Form()], db: Session = Depends(get_db)):
    db_user = crud.get_user_by_username(db=db, username=username)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered.")
    user = schemas.UserCreate(username=username, email=email, password=password)
    return crud.create_user(db=db, user=user)

이 친구가 받아먹고  create user를 해주던지, email 이미 있어서 안된다고 해주던지 하게 된다. 

성공했을 때 홈 화면으로 가고, 실패했으면 안됐다고 alert를 보내주고 여기 머무른다. 적용 화면을 함 보자.

이미 있는 유저 데이터로 했을 때

새 녀석으로 했을 때

DB를 보면 비밀번호도 잘 암호화돼서 저장된 것을 볼 수 있다.

좋다. 이제 로그인 페이지만 후딱 만들어본다.

 

2. 로그인

솔직히 말하면 sign up과 다른 거라곤 데이터 개수와 return값 뿐이다. 바로 jQuery부터 보도록 하겠다. 

<script
  type="text/javascript"
  src="https://cdnjs.cloudflare.com/ajax/libs/jquery-cookie/1.4.1/jquery.cookie.min.js"
></script>
<script type="text/javascript">
  $(document).ready(function ($) {
    $.cookie();
    $(".signin-form").submit(function (event) {
      event.preventDefault();

      var body = new FormData();
      body.append("username", $("#username").val());
      body.append("password", $("#password").val());

      var data = new URLSearchParams(body);

      $.ajax({
        url: "http://localhost:8000/login",
        method: "POST",
        data: data,
        contentType: false,
        processData: false,
        headers: {
          "Content-Type": "application/x-www-form-urlencoded",
        },
        success: function (response) {
          $.cookie("access_token", response.access_token);
          location.href = "http://localhost:8000/dashboard";
        },
        error: function (response) {
          alert(response.responseJSON.detail);
          console.log(response);
        },
      });

      event.preventDefault();
    });

    $("#to-signup").on("click", function () {
      location.href = "http://localhost:8000/signup";
    });
  });
</script>

하나 중요한 것은, cookie 파일이 있어야한다. 

<script
  type="text/javascript"
  src="https://cdnjs.cloudflare.com/ajax/libs/jquery-cookie/1.4.1/jquery.cookie.min.js"
></script>

이런 식으로 삽입을 해주고!!!!!!!! document 로딩이 완료된 후에 cookie 초기화도 해주고!!! 그리고 나서 로그인 함수를 넣어준다. 데이터 만드는 것은 위와 같으니 생략하고 ajax만 보자.

$.ajax({
    url: "http://localhost:8000/login",
    method: "POST",
    data: data,
    contentType: false,
    processData: false,
    headers: {
      "Content-Type": "application/x-www-form-urlencoded",
    },
    success: function (response) {
      $.cookie("access_token", response.access_token);
      location.href = "http://localhost:8000/dashboard";
    },
    error: function (response) {
      alert(response.responseJSON.detail);
      console.log(response);
    },
  });

우선 데이터를 보내주는 url이 달라졌다. login router 주소로 보내주면

@app.post("/login")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: Session = Depends(get_db)):
    user = user_crud.get_user_by_username(db=db, username=form_data.username)
    if user is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect email or password")
    hashed_pass = user.hashed_password
    if not utils.verify_password(form_data.password, hashed_pass):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect email or password")
    
    return {
        "access_token": utils.create_access_token(user.username),
        "refresh_token": utils.create_refresh_token(user.username)
    }

이 친구가 받는다. 싹 처리를 해서 입력 정보가 올바르면 생성된 access token과 refresh token을 받아오고, cookie에 access token을 입력해둔다. 그럼 어떻게 되느냐.

로그인을 성공시키고 F12를 눌러서 개발자 도구 - Application - Cookies를 보면 

Access token이 있다! 이 놈으로 이제 다음 포스팅에 만들 monitoring system의 열람 권한을 확인해줄 것이다. 

토큰이 없다면 이런 내용이 뜨도록! 

하,,, 참 긴 포스팅이었다. 다음 포스팅에서는 Google chart를 활용해서 예쁘게 그래프 그리기... 그리고 열람권한 확인하기를 가지고 돌아오겠다.

틀렸거나 부가적으로 정보를 주고 싶으신 분이 계시다면 언제나 댓글 피드백은 환영입니다!