Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Post 관련 API 구현 #26

Merged
merged 5 commits into from Mar 13, 2019
Merged
Diff settings

Always

Just for now

Copy path View file
@@ -26,7 +26,7 @@ def requested_user(self) -> TblUsers:
user = TblUsers.get_first_or_abort_on_none(
session,
TblUsers.id == get_jwt_identity(),
401
code=401
)

g.requested_user = user
Copy path View file
@@ -11,7 +11,9 @@ def schematics_baseerror_handler(e: BaseError):


def http_exception_handler(e: HTTPException):
return '', e.code
return jsonify({
'msg': e.description
}), e.code


def broad_exception_handler(e: Exception):
Copy path View file
@@ -14,46 +14,86 @@ class Base(_Base):
__abstract__ = True

@classmethod
def get_all(cls, read_session: Session, where_clause: Union[ClauseElement, bool]=True):
def _build_query(
cls, session: Session, where_clause=None, order_by=None
):
query = session.query(cls)

if where_clause is not None:
query = query.filter(where_clause)

if order_by is not None:
query = query.order_by(order_by)

return query

@classmethod
def get_all(
cls, read_session: Session, where_clause: Union[ClauseElement, bool]=None, order_by: ClauseElement=None
):
"""
전달된 session을 통해 cls에 대해 where_clause로 필터해 쿼리하고, .all()의 결과를 반환합니다.
전달된 인자들을 통해 쿼리를 생성하고, .all()의 결과를 반환합니다.
:param read_session: SQLAlchemy session
:param where_clause: 적용할 where clause
:param order_by: 적용할 order by clause
"""
return read_session.query(cls).filter(where_clause).all()
query = cls._build_query(read_session, where_clause, order_by)

return query.all()

@classmethod
def get_first_without_none_check(cls, read_session: Session, where_clause: Union[ClauseElement, bool]=True):
def get_first_without_none_check(
cls, read_session: Session, where_clause: Union[ClauseElement, bool]=True, order_by: ClauseElement=None
):
"""
전달된 session을 통해 cls에 대해 where_clause로 필터해 쿼리하고, None 여부에 상관없이 .first()의 결과를 리턴합니다.
전달된 인자들을 통해 쿼리를 생성하고, None 여부에 상관없이 .first()의 결과를 리턴합니다.
:param read_session: SQLAlchemy session
:param where_clause: 적용할 where clause
:param order_by: 적용할 order by clause
"""
return read_session.query(cls).filter(where_clause).first()
query = cls._build_query(read_session, where_clause, order_by)

return query.first()

@classmethod
def get_first_or_abort_on_none(
cls, read_session: Session, where_clause: Union[ClauseElement, bool]=True,
code: int=404, message: str=None
cls, read_session: Session, where_clause: Union[ClauseElement, bool]=True, order_by: ClauseElement=None,
code: int=404, message: str=None
):
"""
1. 전달된 session을 통해 cls에 대해 where_clause로 필터해 쿼리하고
1. 전달된 인자들을 통해 쿼리를 생성하고
2. .first() 후 결과가 None이면 인자 정보들을 통해 abort,
3. None이 아니라면 해당 객체를 리턴
:param read_session: SQLAlchemy session
:param where_clause: 적용할 where clause
:param order_by: 적용할 order by clause
:param code: .first()의 결과가 None인 경우 abort할 코드
:param message: .first()의 결과가 None인 경우 abort할 때 사용할 메시지
"""
res = cls.get_first_without_none_check(read_session, where_clause)
res = cls.get_first_without_none_check(read_session, where_clause, order_by)

if res is None:
abort(code, message)
else:
return res

@classmethod
def delete(cls, write_session: Session, where_clause: Union[ClauseElement, bool]=True):
def delete(
cls, write_session: Session, where_clause: Union[ClauseElement, bool]=True
):
"""
전달된 session을 통해 cls에 대해 where_clause로 필터해 쿼리하고, 결과를 모두 delete합니다.
commit을 수행하지 않음에 주의하기 바랍니다.
"""
write_session.query(cls).filter(where_clause).delete()

@classmethod
def delete_and_commit(cls, write_session: Session, where_clause: Union[ClauseElement, bool]=True):
def delete_and_commit(
cls, write_session: Session, where_clause: Union[ClauseElement, bool]=True
):
"""
where_clause를 통해 delete 후, commit까지 수행합니다.
"""
Copy path View file
@@ -1,7 +1,9 @@
from flask import abort
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

from app.models import Base
from app.models.user import TblUsers


class TblPosts(Base):
@@ -15,3 +17,51 @@ class TblPosts(Base):

owner = relationship('TblUsers')
category = relationship('TblCategories')

@property
def json_for_list(self):
return {
'id': self.id,
'title': self.title,
'ownerNickname': self.owner.nickname
}

@property
def json_for_detail(self):
return {
'content': self.content
}

@classmethod
def get_post_object_through_id(cls, read_session, id: int) -> 'TblPosts':
post = cls.get_first_or_abort_on_none(read_session, cls.id == id)

return post

@classmethod
def get_post_object_through_id_with_permission_check(
cls, read_session, id: int, requested_user: TblUsers
) -> 'TblPosts':
post = cls.get_post_object_through_id(read_session, id)

if post.owner_id != requested_user.id:
abort(403)

return post

@classmethod
def update_post_with_permission_check(cls, write_session, id: int, requested_user: TblUsers, title, content):
post = cls.get_post_object_through_id_with_permission_check(write_session, id, requested_user)

post.title = title
post.content = content

write_session.add(post)
write_session.commit()

@classmethod
def delete_post_with_permission_check(cls, write_session, id: int, requested_user: TblUsers):
post = cls.get_post_object_through_id_with_permission_check(write_session, id, requested_user)

write_session.delete(post)
write_session.commit()
Copy path View file
@@ -4,7 +4,7 @@

def route(flask_app: Flask):
from app.views.user.account import auth, check_duplicate, refresh, signup
from app.views.board import category
from app.views.board import category, post

handle_exception_func = flask_app.handle_exception
handle_user_exception_func = flask_app.handle_user_exception
@@ -24,6 +24,8 @@ def route(flask_app: Flask):
api_user__account.add_resource(refresh.RefreshAPI, '/refresh')

api_user__board.add_resource(category.CategoryAPI, '/categories')
api_user__board.add_resource(post.PostAPI, '/posts')
api_user__board.add_resource(post.PostItemAPI, '/posts/<int:post_id>')

# - register blueprint
flask_app.register_blueprint(api_v1_blueprint)
Copy path View file
@@ -15,7 +15,8 @@ class Schema:
class Post(BaseModel):
name = StringType(
serialized_name='name',
required=True
required=True,
max_length=TblCategories.name.type.length
)

@validate_with_schematics(PayloadLocation.JSON, Schema.Post)
Copy path View file
@@ -0,0 +1,130 @@
from flask import abort
from flask_jwt_extended import jwt_required
from schematics.types import IntType, StringType
from sqlalchemy.exc import IntegrityError

from app.context import context_property
from app.decorators.validation import PayloadLocation, BaseModel, validate_with_schematics
from app.extensions import main_db
from app.models.post import TblPosts
from app.views.base import BaseResource


class PostAPI(BaseResource):
class Schema:
class Post(BaseModel):
category_id = IntType(
serialized_name='categoryID',
required=True
)

title = StringType(
serialized_name='title',
required=True,
max_length=TblPosts.title.type.length
)

content = StringType(
serialized_name='content',
required=True,
max_length=TblPosts.content.type.length
)

class Get(BaseModel):
category_id = IntType(
serialized_name='category_id',
required=True
)

@validate_with_schematics(PayloadLocation.JSON, Schema.Post)
@jwt_required
def post(self):
"""
게시글 작성 API
"""

payload: self.Schema.Post = context_property.request_payload_object
session = main_db.session
user = context_property.requested_user

try:
post = TblPosts(
title=payload.title,
content=payload.content,
owner_id=user.id,
category_id=payload.category_id
)

session.add(post)
session.commit()
session.refresh(post)

return {
'id': post.id
}
except IntegrityError:
abort(404, 'category ID `{}` not found.'.format(payload.category_id))

@validate_with_schematics(PayloadLocation.ARGS, Schema.Get)
@jwt_required
def get(self):
"""
게시글 목록 API
"""

session = main_db.session
payload: self.Schema.Get = context_property.request_payload_object

return {
'data': [
post.json_for_list for post in TblPosts.get_all(session, TblPosts.category_id == payload.category_id)
]
}


class PostItemAPI(BaseResource):
class Schema:
class Patch(BaseModel):
title = PostAPI.Schema.Post.title
content = PostAPI.Schema.Post.content

def get(self, post_id):
"""
게시글 내용 조회 API
"""

session = main_db.session

post = TblPosts.get_post_object_through_id(session, post_id)

return {
'data': post.json_for_detail
}

@validate_with_schematics(PayloadLocation.JSON, Schema.Patch)
@jwt_required
def patch(self, post_id):
"""
게시글 수정 API
"""

payload: self.Schema.Patch = context_property.request_payload_object
session = main_db.session
user = context_property.requested_user

TblPosts.update_post_with_permission_check(session, post_id, user, payload.title, payload.content)

return {}

@jwt_required
def delete(self, post_id):
"""
게시글 삭제 API
"""

session = main_db.session
user = context_property.requested_user

TblPosts.delete_post_with_permission_check(session, post_id, user)

return {}
ProTip! Use n and p to navigate between commits in a pull request.
You can’t perform that action at this time.