Python RestAPI 구현

python에 대해 공부하던 중 Web으로 구현하기 편한 Flask에 대해 알게 되었는데 정말 말도 안 된다. 어렇게 적은 코드로 Web서버를 구현할 수 있다니. 물론 나중에는 Django를 사용해야 하겠지만 공부하기에는 Flask만큼 편한 것이 없다고 생각한다.

물론 streamlit, Shiny 같은 반응형 웹 앱을 만들기 쉬운 python 기반 프레임워크가 있지만 다루게 된다면 나중에 소개하겠다.

파이썬 백엔드 프레임워크에는 주로 Django, flask가 잘 알려져있는 것 같고 Tornado, Pyramid, FastAPI 등 여러 프레임워크들이 있으나 너무 많아서 필요하게 되면 공부할 것 같다.

서론이 좀 길었는데 본 편은 Flask를 사용한 RestAPI 구현이다.

Flask RestAPI simple Ex

우선 간단하게 구현해봤다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from flask import Flask
from flask_restx import Api, Resource, fields

app = Flask(__name__)
api = Api(app, version='1.0', title='Login API', description='A simple login API')

# 가장 간단한 형태의 인증을 위한 사용자 데이터
users = {
    "user1": "password1",
    "user2": "password2"
}

# Swagger에서 사용할 모델 정의
login_model = api.model('Login', {
    'username': fields.String(required=True, description='사용자명'),
    'password': fields.String(required=True, description='비밀번호')
})

# 로그인을 처리할 리소스
@api.route('/login')
class Login(Resource):
    @api.expect(login_model)  # Swagger에서 사용할 모델 적용, 입력 기대값
    def post(self):
        # Request Parser를 사용하여 요청 데이터 파싱
        # args = login_parser.parse_args()
        # username = args['username']
        # password = args['password']

        data = api.payload  # Swagger UI에서 전송된 데이터 받기
        username = data.get('username')
        password = data.get('password')

        # 유저가 존재하고, 비밀번호가 일치하면 로그인 성공
        if username in users and users[username] == password:
            return {"message": "로그인 성공"}, 200
        else:
            return {"message": "유효하지 않은 사용자명 또는 비밀번호"}, 401

    def get(self):
        return {"users": users}, 200

if __name__ == '__main__':
    app.run(debug=True)

워낙에 기능이 많아서 천천히 늘려나가야 할 것 같다.

여기서 flask-restx를 사용한 이유는 보다 수월하게 설명하기 위해서이다. 그냥 restapi를 구현하는 것을 목표로 했으나 다른 사람이 보기에도 좀 별로다.

그래서 Swagger라는 rest api를 문서화해주는 도구를 지원해주는 매우 유용한 기능이 있다. 사실 직관적으로 보여주기 편해서 사용하는 것이지 너무 귀찮다.

# Swagger

Swagger

이것 말고도 테스트에 용이한 Advanced REST client 같은 크롬 확장 앱도 있다. 근데 이건 못 생겼다. 근데 이게 테스트에는 더 편할수도 있다.

# Advanced REST client

Advanced REST client

아 근데 이 글이 본격적인 시작이라서 말하는 거지만 필자는 conda를 사용하고 있다.

venv같은 것도 있지만 그냥 컴퓨터에 깔려있어서 사용하고 있다.

쨌든 위의 코드를 실행해보면

입력예시

이런 식으로 VSCODE에는 뜬다. 이것 설명하기 귀찮다.

그냥 http://127.0.0.1:5000 ctrl + 마우스 우클릭하든지 들어가면되는데 이것저것 눌러보면 대충 뭐가 뭔지는 알 텐데 일단 swagger는 직관적이어서 좋다.

입력예시

이렇게 그림처럼 입력을 넣을 수도 있고

실패

실패하면 실패 결과 보여주고

실패

성공하면 성공 결과 보여주고 물론 코드를 짜서 그런거지만 매우 보기 편하다.

근데 REST API method에 해당하지 않는 method들은 swagger UI에 추가되지 않는다.

1
2
3
4
5
6
@resource1_api.route('/resource1')
class FirstResource(Resource):
    @resource1_api.expect(parser)
    def test_func(self): # swagger UI에 method가 추가되지 않는다.
        result = {'result_msg': 'Success'}
        return result, 200

파일 분리, 모듈화?

이 부분에 대해서 필자는 매우 중요하다고 생각된다. 웹 개발을 하다보면 코드가 길어지는 경우가 많은데 하나의 파일에 데이터관리도 넣고, 모델도 넣고, 엔드포인트도 설정하고, 비즈니스 로직도 넣어버리면 나중에 코드가 너무 길어져서 뭐가 어디있는지 찾기도 해야하고 다른사람이 봤을 때 매우 지저분 해진다.

이를 위한 것이 있으니 바로 Namespace이다.

Flask자체에도 Blueprint라는 기능이 있긴한데 restx에 있길래 그냥 써보는 것이다.

1
2
3
4
5
6
7
8
9
10
11
12
app/
│
├── controllers/
│   ├── __init__.py
│   └── controller.py
│
├── models/
│   ├── __init__.py
│   └── model.py
│
└── __init__.py
run.py

이런 형태로 이루어져 있다.

/run.py

1
2
3
4
5
6
7
from app import create_app

app = create_app()


if __name__ == '__main__':
    app.run(debug=True)

/app/init.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from flask import Flask
from flask_restx import Api
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
api = Api()

def create_app():
  
  app = Flask(__name__)
  app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///api.db'

  db.init_app(app)
  api.init_app(app)
  
  from app.controllers.controller import ns
  api.add_namespace(ns, "/Testing")

  return app

/app/controllers/controller.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from flask_restx import Resource, fields, Namespace
from app import api, db
from app.models.model import User

ns = Namespace("Login test", description="simple login")

# Swagger에서 사용할 모델 정의
login_model = api.model('Login', {
    'username': fields.String(required=True, description='사용자명', example="cl_victor"),
    'password': fields.String(required=True, description='비밀번호', example="1234")
})

# 사용자 수정을 위한 모델
update_model = api.model('Update', {
    'username': fields.String(required=True, description='사용자명'),
    'password': fields.String(required=True, description='새로운 비밀번호')
})

# 로그인을 처리할 리소스
@ns.route('/login')
class Login(Resource):
    @api.expect(login_model)  # Swagger에서 사용할 모델 적용
    def post(self):
        data = api.payload  # Swagger UI에서 전송된 데이터 받기
        username = data.get('username')
        password = data.get('password')

        # 데이터베이스에 사용자 추가
        new_user = User(username=username, password=password)
        db.session.add(new_user)
        db.session.commit()

        return {"message": "사용자가 추가되었습니다."}, 201

    def get(self):
        # 모든 사용자 가져오기
        users = User.query.all()
        user_list = [{"id": user.id, "username": user.username, "password": user.password} for user in users]
        return {"users": user_list}, 200
    
# 사용자 수정을 위한 리소스
@ns.route('/login/<string:username>')
class UpdateUser(Resource):
    def get(self, username):
        user = User.query.filter_by(username=username).first()
        if not user:
            return {"message": "사용자를 찾을 수 없습니다."}, 404

        return {"id": user.id, 
                "username": user.username, 
                "password": user.password}, 200
    
    @api.expect(update_model)
    def put(self, username):
        data = api.payload
        new_username = data.get('username')
        new_password = data.get('password')
        
        user = User.query.filter_by(username=username).first()
        if not user:
            return {"message": "사용자를 찾을 수 없습니다."}, 404

        user.username = new_username
        user.password = new_password
        db.session.commit()

        return {"message": "사용자 정보가 수정되었습니다.",
                "user-info": {
                    "username": user.username, "password": user.password
                    }}, 200
        
    def delete(self, username):
        user = User.query.filter_by(username=username).first()
        if not user:
            return {"message": "사용자를 찾을 수 없습니다."}, 404

        db.session.delete(user)
        db.session.commit()

        return {"message": "사용자가 삭제되었습니다."}, 200

/app/models/model.py

1
2
3
4
5
6
7
from app import db

# 데이터베이스 모델 정의
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    password = db.Column(db.String(50), nullable=False)

이미 인터넷에 많은 코드가 돌아다니기 때문에 그쪽을 참고 하는 것도 좋다.

add_namespace()

외부에서 클래스를 구현후 add_namespace()를 통해 클래스를 등록

1
2
3
4
5
6
7
8
9
10
11
12
13
14
namespace = Namespace('hello')  # 첫 번째

@namespace.route('/')
class HelloWorld(Resource):  
    def get(self):
        return {"hello" : "world!"}, 201, {"hi":"Namespace"}
    
api.add_namespace(namespace, '/hello2')  


@api.route('/hello')  # 두 번째
class HelloWorld(Resource):
    def get(self):
        return {"hello" : "world!"}, 201, {"hi":"hello"}

자세한 내용은 Flask-RESTX 공식 문서 참조