Flask로 API 서버 만들기 (5) - User Operations

Developing API Sample Server by Flask 

Original Post: How to structure a Flask-RESTPlus web service for production builds

Github: https://github.com/cosmic-byte/flask-restplus-boilerplate

Flask로 API 서버 만들기 (1) - 개발 환경 준비

Flask로 API 서버 만들기 (2) - config 와 실행 확인

Flask로 API 서버 만들기 (3) - User 테이블 만들기

Flask로 API 서버 만들기 (4) - Testing

Flask로 API 서버 만들기 (5) - User Operations

Flask로 API 서버 만들기 (6) - Security and Authentication

Flask로 API 서버 만들기 (7) - Route protection and Authorization

Flask로 API 서버 만들기 (8) - Extra tips (Makefiles)

User 테이블을 만들었으니 User 처리에 대한 기능을 만들어 보겠습니다.

만들 기능은 아래와 같습니다.

1. 새로운 User 생성

2. 등록된 User의 public_id 확인하기

3. 등록된 모든 User 불러오기

service 폴더 안에 user_service.py 파일을 만들어 줍니다.

$ sudo vim ./app/main/service/user_service.py

user_service.py 파일안에 아래 script를 추가합니다.

import uuid
import datetime

from app.main import db
from app.main.model.user import User


def save_new_user(data):
    user = User.query.filter_by(email=data['email']).first()
    if not user:
        new_user = User(
            public_id=str(uuid.uuid4()),
            email=data['email'],
            username=data['username'],
            password=data['password'],
            registered_on=datetime.datetime.utcnow()
        )
        save_changes(new_user)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.'
        }
        return response_object, 201
    else:
        response_object = {
            'status': 'fail',
            'message': 'User already exists. Please Log in.',
        }
        return response_object, 409


def get_all_users():
    return User.query.all()


def get_a_user(public_id):
    return User.query.filter_by(public_id=public_id).first()


def save_changes(data):
    db.session.add(data)
    db.session.commit()

save_new_user 함수는 신규 유저를 추가하는 기능입니다.

먼저 email이 같은 유저가 있는지를 확인합니다. 없다면 새로운 User를 추가합니다.

get_all_users 함수는 모든 User 정보를 불러오는 기능입니다.

get_a_user 함수는 public_id를 조회하고 매칭되는 User 정보를 불어옵니다.

save_changes 함수는 User 데이터를 저장(업데이트)하는 함수입니다.

이번에는 Utility 기능들을 만들어 보겠습니다.

main 폴더 아래 util 폴더를 만들고 dto.py 파일을 만들어 줍니다.

dto는 data transfer object로 데이터를 전달하는 기능을 합니다.

$ mkdir ./app/main/util
$ sudo vim ./app/main/util/dto.py

dto.py에 User 데이터를 전달하는 함수를 만들어 줍니다.

from flask_restplus import Namespace, fields


class UserDto:
    api = Namespace('user', description='user related operations')
    user = api.model('user', {
        'email': fields.String(required=True, description='user email address'),
        'username': fields.String(required=True, description='user username'),
        'password': fields.String(required=True, description='user password'),
        'public_id': fields.String(description='user Identifier')
    })

user라는 Namespace를 만들고 api 변수에 할당합니다.

그리고 api.model을 이용해 user 데이터를 담을 필드를 만들어 줍니다.

이런게 하면 Flask 안에서 User 데이터를 담에 전달할 수 있습니다.

이제 User 데이터를 http로 처리할 controller를 만들어 주겠습니다.

controller 폴더 안에 user_controller.py 파일을 만들어 줍니다.

$ sudo vim ./app/main/controller/user_controller.py

user_controller.py 파일안에 아래 script를 추가합니다.

from flask import request
from flask_restplus import Resource

from ..util.dto import UserDto
from ..service.user_service import save_new_user, get_all_users, get_a_user

api = UserDto.api
_user = UserDto.user


@api.route('/')
class UserList(Resource):
    @api.doc('list_of_registered_users')
    @api.marshal_list_with(_user, envelope='data')
    def get(self):
        """List all registered users"""
        return get_all_users()

    @api.response(201, 'User successfully created.')
    @api.doc('create a new user')
    @api.expect(_user, validate=True)
    def post(self):
        """Creates a new User """
        data = request.json
        return save_new_user(data=data)


@api.route('/')
@api.param('public_id', 'The User identifier')
@api.response(404, 'User not found.')
class User(Resource):
    @api.doc('get a user')
    @api.marshal_with(_user)
    def get(self, public_id):
        """get a user given its identifier"""
        user = get_a_user(public_id)
        if not user:
            api.abort(404)
        else:
            return user

app 폴더 밑에있는 __init__.py 파일안에 API Server의 entry point를 잡아줍니다.

$ sudo vim ./app/__init__.py

__init__.py파일 안에 아래와 같이 추가합니다.

# app/__init__.py

from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')

manage.py 파일 안에 blueprint를 추가해 줍니다.

$ sudo vim ./manage.py

추가할 내용은 blueprint 모듈 추가(from app import blueprint)와 app.register_blueprint(blueprint) 입니다.

import os
import unittest

from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager

from app import blueprint
from app.main import create_app, db
from app.main.model import user

app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')
app.register_blueprint(blueprint)

app.app_context().push()

manager = Manager(app)

migrate = Migrate(app, db)

manager.add_command('db', MigrateCommand)

@manager.command
def run():
        app.run()

@manager.command
def test():
        """Runs the unit tests."""
        tests = unittest.TestLoader().discover('app/test', pattern='test*.py')
        result = unittest.TextTestRunner(verbosity=2).run(tests)
        if result.wasSuccessful():
                return 0
        return 1

if __name__ == '__main__':
        manager.run()

이제 manage.py로 API 서버를 실행합니다.

$ python manage.py run
 * Serving Flask app "app.main" (lazy loading)
 * Environment: development
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 145-484-411

브라우저를 열고 http://localhost:5000으로 접속합니다.

만약 Flask Server가 원격으로 있다면 외부에서 접근을 허용해야 합니다.

manage.py 안에 app.run()을 호출할때 host='0.0.0.0'을 추가합니다.

app.run(host='0.0.0.0')

그리고 다시 실행하면 Running on 에 아래와 같이 나오게 되며 외부에서도 접속이 가능해 집니다.

* Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)

$ python manage.py run
 * Serving Flask app "app.main" (lazy loading)
 * Environment: development
 * Debug mode: on
 * Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 145-484-411

다시 브라우저를 열고 내부 IP와 port 5000으로 접속합니다.

POST /user 텝을 열고 Try it out 버튼을 클릭한 후 아래와 같이 email, username, password, public_id을 json 타입에 맞게 입력하고 Execute를 클릭합니다.

그럼 아래와 같이 결과가 나옵니다.

정상적으로 등록되었는지 GET /user/를 클릭해 보겠습니다.

GET /user 텝을 열고 Try it out 버튼을 클릭하고 바로 Execute 버튼을 클릭합니다.

그럼 등록한 tongchun의 정보가 보입니다.

여기까지 User 테이블을 만들고 User_Coltroller와 blueprint를 이용해 swagger를 실행해 데이터를 입력하고 출력하는 것을 확인했습니다.