INDEX

이번 포스트는 macOS Mojave 10.14.5 에서 수행된 작업입니다.

개발 환경

MacBook Pro (13-inch, 2017, Four Thunderbolt 3 Ports)
Python 3.7
vscode
Docker-Compose version 1.23.2, build 1110ad01

RESTPlus 라이브러리를 사용해 REST API 만들기(Users Part)

flask-restplus 종속성을 설치합니다

1	$ pipenv install flask-restplus

이제 app에 설치한 RESTPlus를 달아 봅시다

# ./app/__init__.py

"""
    APP을 실행하기 위해 config file
"""

from flask import Flask
from app.api.database import DB, MA
from app.api import REST_API
from app.constants import SQLALCHEMY_DATABASE_URI_FORMAT


def create_app()->(Flask):
    """ create_app()을 호출하여 app을 초기화 """
    app = Flask(__name__)
    app.app_context().push()
    
    app.config['SQLALCHEMY_DATABASE_URI'] = SQLALCHEMY_DATABASE_URI_FORMAT
    app.config['SQLALCHEMY_ECHO'] = True
    app.config['DEBUG'] = True

    DB.init_app(app)
    REST_API.init_app(app)
    MA.init_app(app)
    
    return app

MVC 패턴을 이용해 User 부분을 작성합니다

# ./app/users/views.py

"""
    User views file
"""
from http import HTTPStatus
from flask import jsonify, make_response
from sqlalchemy.exc import SQLAlchemyError
from flask_restplus import Namespace, Resource, reqparse, fields
from app.users.models import Users, UsersSchema
from app.api.database import DB

API = Namespace('Users', description="User's RESTPlus - API")
USERS_SCHEMA = UsersSchema()

@API.route('s')
class UsersAuth(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('user_id', required=True, type=str, help="User's ID", location='json')
    parser.add_argument('user_password', required=True, type=str, help="User's PW", location='json')
    parser.add_argument('user_email', required=True, type=str, help="User's Email", location='json')

    users_field = API.model('Sign up', {
        'user_id' : fields.String,
        'user_password' : fields.String,
        'user_email' : fields.String
    })

    @API.doc('post')
    @API.expect(users_field)
    def post(self):
        args_ = self.parser.parse_args()
        user = Users(args_['user_id'], args_['user_password'], args_['user_email'])
        try:
            DB.session.add(user)
            DB.session.commit()
            body = jsonify({'users' : USERS_SCHEMA.dump(user).data})
            code = HTTPStatus.OK
        except SQLAlchemyError as err:
            DB.session.rollback()
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

@API.route('/auth')
class UserAuth(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('user_id', required=True, type=str, help="User's ID", location='json')
    parser.add_argument('user_password', required=True, type=str, help="User's PW", location='json')

    user_login_field = API.model('Sign in', {
        'user_id' : fields.String,
        'user_password' : fields.String
    })

    @API.doc('post')
    @API.expect(user_login_field)
    def post(self):
        args_ = self.parser.parse_args()
        try:
            user = Users.query.filter(Users.user_id == args_['user_id']).first()
            body = jsonify({'user_id' : user.user_id})
            code = HTTPStatus.OK
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

Swagger를 달아 API를 호출 할수있는 Docs를 만듭시다

# ./app/api/**init**.py

from flask_restplus import Api
from app.users.views import API as users_api

REST_API = Api()

REST_API.add_namespace(users_api, '/user')

앱을 한번 실행해 봅시다

1	$ python manage.py run

RESTPlus API 만들기(Posts Part)

이전 포스트와 같이 Posts에 대한 Model을 만듭니다

# ./app/posts/models.py

"""
    Posts model file
"""

from app.api.database import DB, MA
from marshmallow import Schema, fields, validate
from app.users.models import Users, UsersSchema
from sqlalchemy.sql import text

class Posts(DB.Model):
    __tablename__ = 'posts'
    __table_args__ = {'mysql_collate': 'utf8_general_ci'}

    id = DB.Column(DB.Integer, primary_key=True)
    author_id = DB.Column(DB.String(255), DB.ForeignKey(Users.user_id))
    title = DB.Column(DB.String(512), nullable=False)
    body = DB.Column(DB.String(1024), nullable=False)
    author = DB.relationship('Users', uselist=False)
    created = DB.Column(DB.TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), nullable=False)

    def __init__(self, author_id, title, body):
        self.author_id = author_id
        self.title = title
        self.body = body

class PostsSchema(MA.Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer()
    author_id = fields.String(validate=not_blank)
    title = fields.String(validate=not_blank)
    body = fields.String(validate=not_blank)
    author = fields.Nested(UsersSchema)
    created = fields.String(validate=not_blank)

MVC 패턴을 이용해 Posts 부분을 작성합니다

# ./app/posts/views.py

"""
    Posts view file
"""

from flask_restplus import Namespace, Resource, reqparse, fields
from flask import jsonify, make_response
from http import HTTPStatus
from sqlalchemy.exc import SQLAlchemyError
from app.posts.models import Posts, PostsSchema
from app.users.models import Users, UsersSchema
from app.api.database import DB

API = Namespace('Posts', description="Post's REST API")
POSTS_SCHEMA = PostsSchema()

@API.route('s')
class Post(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('author_id', required=True, type=str, help="Post's author ID", location='json')
    parser.add_argument('title', required=True, type=str, help="Post's title", location='json')
    parser.add_argument('body', required=True, type=str, help="Post's body", location='json')

    post_field = API.model('Post', {
        'author_id': fields.String,
        'title': fields.String,
        'body': fields.String
    })

    @API.doc('get')
    def get(self):
        try:
            posts = Posts.query.all()
            body = jsonify(POSTS_SCHEMA.dump(posts, many=True).data)
            code = HTTPStatus.OK
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

    @API.expect(post_field)
    @API.doc('post')
    def post(self):
        args_ = self.parser.parse_args()
        post = Posts(author_id=args_['author_id'], title=args_['title'], body=args_['body'])
        try:
            DB.session.add(post)
            DB.session.commit()
            body = jsonify({'post', POSTS_SCHEMA.dump(post).data})
            code = HTTPStatus.OK
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

@API.route('/<int:reqno>')
class PostItem(Resource):
    def get(self, reqno):
        try:
            post = DB.session.query(Posts).outerjoin(
                Users, Users.user_id == Posts.author_id).filter(Posts.id==reqno).first()
            body = jsonify({'post' : POSTS_SCHEMA.dump(post).data})
            code = HTTPStatus.OK
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

Swagger를 달아 API를 호출 할수있는 Docs를 만듭시다

# ./app/api/__init__.py
"""
    API config file
"""

from flask_restplus import Api
from app.posts.views import API as posts_api

REST_API = Api()

REST_API.add_namespace(users_api, '/user')
REST_API.add_namespace(posts_api, '/post')

MVC(Model, View, Controller) 분리

# ./app/api/database.py

"""
    Create db
"""
from flask import jsonify
from flask import make_response
from http import HTTPStatus
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import IntegrityError
from flask_marshmallow import Marshmallow

DB = SQLAlchemy()
MA = Marshmallow()

class CRUD:
    body = ''
    status_code = HTTPStatus.NOT_IMPLEMENTED

    def add(self, resource, schema):
        try:
            DB.session.add(resource)
            DB.session.commit()
            self.body = jsonify(schema.dump(resource).data)
            self.status_code = HTTPStatus.OK
        except IntegrityError as err:
            DB.session.rollback()
            err_meg = str(err)
            self.body = jsonify({'error' : err_meg, 'type' : 'IntegrityError'})
            if "Duplicate entry" in err_meg:
                self.status_code = HTTPStatus.CONFLICT
            else:
                self.status_code = HTTPStatus.BAD_REQUEST
        return make_response(self.body, self.status_code)

User, Post Model을 따로 빼 줍니다

# ./app/users/models.py

"""
    Users models file
"""
from sqlalchemy.sql import text
from app.api.database import DB, MA, CRUD
from flask_sqlalchemy import SQLAlchemy
from marshmallow import Schema, fields, validate

class Users(DB.Model, CRUD):
    __tablename__ = 'users'
    __table_args__ = {'mysql_collate': 'utf8_general_ci'}

    id = DB.Column(DB.Integer, primary_key=True)
    user_id = DB.Column(DB.String(255), unique=True, nullable=False)
    user_password = DB.Column(DB.String(255), nullable=False)
    user_email = DB.Column(DB.String(255), nullable=False)
    created = DB.Column(DB.TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), nullable=False)

    def __init__(self, user_id : str, user_password : str, user_email : str):
        self.user_id = user_id
        self.user_password = user_password
        self.user_email = user_email

class UsersSchema(MA.Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer(dump_only=True)
    user_id = fields.String(validate=not_blank)
    user_password = fields.String(validate=not_blank)
    user_email = fields.String(validate=not_blank)
    created = fields.String(validate=not_blank)

# ./app/posts/models.py

"""
    Posts model file
"""

from app.api.database import DB, MA, CRUD
from marshmallow import Schema, fields, validate
from app.users.models import Users, UsersSchema
from sqlalchemy.sql import text

class Posts(DB.Model, CRUD):
    __tablename__ = 'posts'
    __table_args__ = {'mysql_collate': 'utf8_general_ci'}

    id = DB.Column(DB.Integer, primary_key=True)
    author_id = DB.Column(DB.String(255), DB.ForeignKey(Users.user_id))
    title = DB.Column(DB.String(512), nullable=False)
    body = DB.Column(DB.String(1024), nullable=False)
    author = DB.relationship('Users', uselist=False)
    created = DB.Column(DB.TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), nullable=False)

    def __init__(self, author_id, title, body):
        self.author_id = author_id
        self.title = title
        self.body = body

class PostsSchema(MA.Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer()
    author_id = fields.String(validate=not_blank)
    title = fields.String(validate=not_blank)
    body = fields.String(validate=not_blank)
    author = fields.Nested(UsersSchema)
    created = fields.String(validate=not_blank)

view 부분도 빼 줍시다

# ./app/users/views.py

"""
    User views file
"""
from http import HTTPStatus
from flask import jsonify, make_response
from sqlalchemy.exc import SQLAlchemyError
from flask_restplus import Namespace, Resource, reqparse, fields
from app.users.models import Users, UsersSchema
from app.api.database import DB

API = Namespace('Users', description="User's RESTPlus - API")
USERS_SCHEMA = UsersSchema()

@API.route('s')
class UsersAuth(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('user_id', required=True, type=str, help="User's ID", location='json')
    parser.add_argument('user_password', required=True, type=str, help="User's PW", location='json')
    parser.add_argument('user_email', required=True, type=str, help="User's Email", location='json')

    users_field = API.model('userRegister', {
        'user_id' : fields.String,
        'user_password' : fields.String,
        'user_email' : fields.String
    })

    @API.doc('post')
    @API.expect(users_field)
    def post(self):
        args_ = self.parser.parse_args()
        user = Users(args_['user_id'], args_['user_password'], args_['user_email'])
        return user.add(user, USERS_SCHEMA)

@API.route('/auth')
class UserAuth(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('user_id', required=True, type=str, help="User's ID", location='json')
    parser.add_argument('user_password', required=True, type=str, help="User's PW", location='json')

    user_login_field = API.model('userLogin', {
        'user_id' : fields.String,
        'user_password' : fields.String
    })

    @API.doc('post')
    @API.expect(user_login_field)
    def post(self):
        args_ = self.parser.parse_args()
        try:
            user = Users.query.filter(Users.user_id == args_['user_id'], Users.user_password  == args_['user_password']).first()
            body = jsonify({'user_id' : user.user_id})
            if user:
                code = HTTPStatus.OK
            else:
                code = HTTPStatus.NOT_FOUND
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

# ./app/posts/views.py

"""
    Posts view file
"""

from flask_restplus import Namespace, Resource, reqparse, fields
from flask import jsonify, make_response
from http import HTTPStatus
from sqlalchemy.exc import SQLAlchemyError
from app.posts.models import Posts, PostsSchema
from app.users.models import Users, UsersSchema
from app.api.database import DB

API = Namespace('Posts', description="Post's REST API")
POSTS_SCHEMA = PostsSchema()

@API.route('s')
class Post(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('author_id', required=True, type=str, help="Post's author ID", location='json')
    parser.add_argument('title', required=True, type=str, help="Post's title", location='json')
    parser.add_argument('body', required=True, type=str, help="Post's body", location='json')

    post_field = API.model('Post', {
        'author_id': fields.String,
        'title': fields.String,
        'body': fields.String
    })

    @API.doc('get')
    def get(self):
        try:
            posts = Posts.query.all()
            body = jsonify(POSTS_SCHEMA.dump(posts, many=True).data)
            if posts:
                code = HTTPStatus.OK
            else:
                code = HTTPStatus.NOT_FOUND
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

    @API.expect(post_field)
    @API.doc('post')
    def post(self):
        args_ = self.parser.parse_args()
        post = Posts(author_id=args_['author_id'], title=args_['title'], body=args_['body'])
        return post.add(post, POSTS_SCHEMA)

@API.route('/<int:reqno>')
class PostItem(Resource):
    def get(self, reqno):
        try:
            post = DB.session.query(Posts).outerjoin(
                Users, Users.user_id == Posts.author_id).filter(Posts.id==reqno).first()
            body = POSTS_SCHEMA.dump(post).data
            if post:
                code = HTTPStatus.OK
            else:
                code = HTTPStatus.NOT_FOUND
        except SQLAlchemyError as err:
            body = jsonify({'message' : str(err)})
            code = HTTPStatus.INTERNAL_SERVER_ERROR
        return make_response(body, code.value)

이제 최종적으로 실행을 해봅시다

1	$ python manage.py run

Flask REST API CRUD Board 만들기 MVC 그리고 REST API Swagger(4) - [Backend Framework]

INDEX

개발 환경

RESTPlus 라이브러리를 사용해 REST API 만들기(Users Part)

RESTPlus API 만들기(Posts Part)

MVC(Model, View, Controller) 분리

INDEX

Myeong Seok KIM

Flask REST API CRUD Board 만들기 MVC 그리고 REST API Swagger(4) - [Backend Framework]

INDEX

개발 환경

RESTPlus 라이브러리를 사용해 REST API 만들기(Users Part)

RESTPlus API 만들기(Posts Part)

MVC(Model, View, Controller) 분리

INDEX

Myeong Seok KIM

Mac에서 멋진 터미널 만들기

멋진 맥 터미널 만들기!

Docker Compose 설치하기

Docker Compose 설치하기

AWS? AWS가 뭐야? - [Cloud Infra]

AWS가 뭐야?

AWS EC2에 대해서 알아보기 - [Cloud Infra]

AWS가 뭐야?

Flask REST API CRUD Board 만들기 실행(1) - [Backend Framework]

첫번째 단계

Flask REST API CRUD Board 만들기 설정 및 MySQL(2) - [Backend Framework]

두번째 단계

Flask REST API CRUD Board ORM 그리고 Model(3) - [Backend Framework]

세번째 단계

Flask REST API CRUD Board 만들기 MVC 그리고 REST API Swagger(4) - [Backend Framework]

네번째 단계

Flask REST API CRUD Board Auth 그리고 암호화(5) - [Backend Framework]

다섯번째 단계

Hexo Blog 만들기

Github Hexo Blog 만들어 보기