Commit d970577b authored by 박철완's avatar 박철완

initial commit

parents
ENV=dev
DEBUG=true
APP_HOST=127.0.0.1
APP_PORT=5000
DB_HOST=127.0.0.1
DB_PORT=5432
DB_AUTHEN=server:asdf1234 # username:password 형식
DB_NAME=boilerplate
JWT_SECRET_KEY=secret
\ No newline at end of file
# Fastapi boilerplate
```
# for development
uvicorn app:app --host 127.0.0.1 --port <port> --reload --workers <N>
# for production
gunicorn app:app --workers <N> --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:<port>
```
```
# create new revision
alembic revision --autogenerate -m "description of new revision"
```
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "version_path_separator" below.
# version_locations = %(here)s/bar:%(here)s/bat:migrations/versions
# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
; sqlalchemy.url = driver://user:pass@localhost/dbname
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
from fastapi import FastAPI, Request, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from core.exception import CustomException
from core.config import config
from core.fastapi.auth.middleware import AuthBackend, AuthMiddleware, auth_error_handler
from core.fastapi.session import SessionMiddleware
from core.db import disconnect, session
from core.db.session import get_session_id
from app.routes import init_routers
def init_cors(app: FastAPI):
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
def init_handlers(app: FastAPI):
@app.exception_handler(CustomException)
async def exception_handler(request: Request, exception: CustomException):
return JSONResponse(
status_code=exception.code,
content={
'message': exception.message
}
)
def init_middlewares(app: FastAPI):
app.add_middleware(SessionMiddleware)
app.add_middleware(
AuthMiddleware,
backend=AuthBackend(),
on_error=auth_error_handler,
)
def init_event_handlers(app: FastAPI):
@app.on_event('startup')
def on_startup():
pass
@app.on_event('shutdown')
def on_shutdown():
disconnect()
def create_app() -> FastAPI:
app = FastAPI(
title=config.APP_TITLE,
description=config.APP_DESCRIPTION,
version=config.APP_VERSION,
docs_url='/docs' if config.ENV == 'dev' else None,
redoc_url='/redoc' if config.ENV == 'dev' else None,
dependencies=[]
)
init_cors(app)
init_routers(app)
init_handlers(app)
init_middlewares(app)
init_event_handlers(app)
return app
app = create_app()
@app.get('/')
def healthy():
return JSONResponse(
status_code=200,
content={
'msg': 'healthy',
'session': str(session),
'session_id': get_session_id()
}
)
__all__ = [
'app'
]
\ No newline at end of file
from core.exception import NotFoundException, ForbiddenException
class MemoNotFoundException(NotFoundException):
message = 'there is no memo with given info'
class NotAnOwnerException(ForbiddenException):
message = 'owner permission is required'
__all__ = [
'MemoNotFoundException',
'NotAnOwnerException'
]
\ No newline at end of file
from core.exception import BadRequestException, NotFoundException, ForbiddenException
class PasswordDoesNotMatchException(BadRequestException):
message = 'given two passwords do not match'
class DuplicatedUserInfoException(BadRequestException):
message = 'duplicated user info is in the database'
class UserNotFoundException(NotFoundException):
message = 'there is no user with given info'
class IncorrectPasswordException(BadRequestException):
message = 'given password is incorrect'
class NotAnAdminException(ForbiddenException):
message = 'admin permission is required'
__all__ = [
'PasswordDoesNotMatchException',
'DuplicatedUserInfoException',
'UserNotFoundException',
'IncorrectPasswordException',
'NotAnAdminException'
]
\ No newline at end of file
from .user import User
__all__ = [
'User',
]
\ No newline at end of file
from sqlalchemy import VARCHAR, BigInteger, Column, ForeignKey, Unicode
from sqlalchemy.orm import relationship
from core.db import ModelBase, TimestampMixin
class Memo(ModelBase, TimestampMixin):
id = Column(BigInteger, primary_key=True, autoincrement=True, index=True)
title = Column(Unicode(255), nullable=False)
content = Column(VARCHAR(length=64), nullable=False)
writer = relationship('User', back_populates='memos')
writer_id = Column(BigInteger, ForeignKey('users.id'))
__all__ = [
'Memo'
]
\ No newline at end of file
from xmlrpc.client import Boolean
from sqlalchemy import BigInteger, Column, Unicode, Boolean
from sqlalchemy.orm import relationship
from core.db import ModelBase, TimestampMixin
class User(ModelBase, TimestampMixin):
id = Column(BigInteger, primary_key=True, autoincrement=True, index=True)
password = Column(Unicode(255), nullable=False)
email = Column(Unicode(255), nullable=False, unique=True)
nickname = Column(Unicode(255), nullable=False, unique=True)
is_admin = Column(Boolean, default=False)
memos = relationship("Memo", back_populates='writer')
__all__ = [
'User'
]
\ No newline at end of file
from fastapi import Request
from app.services.memo import MemoService
from app.exceptions.memo import MemoNotFoundException, NotAnOwnerException
from core.fastapi.permission import BasePermission
class IsMemoOwner(BasePermission):
exception = NotAnOwnerException
async def has_permission(self, request: Request) -> bool:
id = request.path_params.get('id')
if id is None:
return False
memo = await MemoService.find(id=id)
if memo is None:
raise MemoNotFoundException
return memo.writer_id == request.user.id
__all__ = [
'IsMemoOwner'
]
\ No newline at end of file
from fastapi import Request
from app.exceptions.user import NotAnAdminException
from core.fastapi.permission import BasePermission
from core.exception import UnauthorizedException
from app.services.users import UserService
class IsAuthenticated(BasePermission):
exception = UnauthorizedException
async def has_permission(self, request: Request) -> bool:
return request.user.id is not None
class IsAdmin(BasePermission):
exception = NotAnAdminException
async def has_permission(self, request: Request) -> bool:
if request.user.id is None:
return False
user = await UserService.find(id=request.user.id)
return user is not None and user.is_admin
__all__ = [
'IsAuthenticated',
'IsAdmin'
]
\ No newline at end of file
from fastapi import FastAPI
from .user import user_router
from .memo import memo_router
def init_routers(app: FastAPI):
app.include_router(user_router, prefix='/user', tags=['User'])
app.include_router(memo_router, prefix='/memo', tags=['Memo'])
__all__ = [
'init_routers'
]
\ No newline at end of file
from typing import *
from fastapi import APIRouter, Depends, Request
from app.permissions.memo import IsMemoOwner
from app.services.memo import MemoService
from app.views.memo import MemoRequestView, MemoResponseView, PatchMemoRequestView
from app.permissions.user import IsAuthenticated, IsAdmin
from core.fastapi.permission import PermissionDependency
from fastapi_pagination import Page, paginate, Params
memo_router = APIRouter()
@memo_router.get(
'/',
response_model=Page[MemoResponseView],
dependencies=[Depends(PermissionDependency(IsAuthenticated))]
)
async def get_memos(request: Request, pagination_params: Params = Depends()):
return paginate(await MemoService.find_all(writer_id=request.user.id), params=pagination_params)
@memo_router.post(
'/',
response_model=MemoResponseView,
dependencies=[Depends(PermissionDependency(IsAuthenticated))]
)
async def create_memo(body: MemoRequestView, request: Request):
return await MemoService.create(writer_id=request.user.id, **body.dict())
@memo_router.get(
'/{id}',
response_model=MemoResponseView,
dependencies=[Depends(PermissionDependency(IsAuthenticated & (IsAdmin | IsMemoOwner)))]
)
async def get_memo(id: int):
return await MemoService.find(id=id)
@memo_router.patch(
'/{id}',
response_model=MemoResponseView,
dependencies=[Depends(PermissionDependency(IsAuthenticated & (IsAdmin | IsMemoOwner)))]
)
async def patch_memo(id: int, body: PatchMemoRequestView):
return await MemoService.patch(memo_id=id, **body.dict())
@memo_router.delete(
'/{id}',
dependencies=[Depends(PermissionDependency(IsAuthenticated & (IsAdmin | IsMemoOwner)))]
)
async def delete_memo(id: int):
await MemoService.delete(memo_id=id)
__all__ = [
'memo_router'
]
\ No newline at end of file
from fastapi import APIRouter, Request, Depends
from app.views.user import CreateUserRequestView, CreateUserResponseView, GetUserResponseView, LoginRequestView, LoginResponseView
from app.services.users import UserService
from app.permissions.user import IsAuthenticated
from core.fastapi.permission import PermissionDependency
user_router = APIRouter()
@user_router.post(
'/',
response_model=CreateUserResponseView,
)
async def create_user(request: CreateUserRequestView):
return await UserService.create(**request.dict())
@user_router.get(
'/',
response_model=GetUserResponseView,
dependencies=[Depends(PermissionDependency(IsAuthenticated))]
)
async def get_user(request: Request):
return await UserService.find(id=request.user.id)
@user_router.post(
'/login',
response_model=LoginResponseView,
)
async def login(request: LoginRequestView):
user, jwt = await UserService.login(**request.dict())
return LoginResponseView(
email=user.email,
nickname=user.nickname,
token=jwt
)
__all__ = [
'user_router'
]
\ No newline at end of file
import sqlalchemy
from app.exceptions.memo import MemoNotFoundException
from app.exceptions.user import UserNotFoundException
from app.models.memo import Memo
from app.models.user import User
from app.services.users import UserService
from core.db import session
from typing import *
from core.db.transaction import Transaction
class MemoService:
@staticmethod
async def find(
id: int | None = None,
title: str | None = None,
) -> Memo | None:
if id is None and title is None:
return None
id_cri = (Memo.id == id) if id is not None else sqlalchemy.true()
title_cri = (Memo.title == title) if title is not None else sqlalchemy.true()
query = session.query(Memo).filter(id_cri & title_cri)
return query.first()
@staticmethod
async def find_all(
writer_id: int | None = None,
writer: User | None = None,
) -> List[Memo]:
if writer is None:
writer = await UserService.find(id=writer_id)
if writer is None:
raise UserNotFoundException
return writer.memos
@staticmethod
async def create(
title: str,
content: str,
writer_id: int,
) -> Memo:
with Transaction():
if await UserService.find(id=writer_id) is None:
raise UserNotFoundException
memo = Memo(
title=title,
content=content,
writer_id=writer_id
)
session.add(memo)
return memo
@staticmethod
async def delete(
memo_id: int | None = None,
memo: Memo | None = None,
):
with Transaction():
if memo is None:
memo = await MemoService.find(id=memo_id)
if memo is None:
raise MemoNotFoundException
session.delete(memo)
@staticmethod
async def patch(
memo_id: int | None = None,
memo: Memo | None = None,
title: str | None = None,
content: str | None = None,
) -> Memo:
with Transaction():
if memo is None:
memo = await MemoService.find(id=memo_id)
if memo is None:
raise MemoNotFoundException
if title is not None:
memo.title = title
if content is not None:
memo.content = content
return memo
__all__ = [
'MemoService'
]
\ No newline at end of file
from typing import *
import sqlalchemy
from app.exceptions.user import DuplicatedUserInfoException, IncorrectPasswordException, PasswordDoesNotMatchException, UserNotFoundException
from app.models.user import User
from core.db import session
from core.db.transaction import Transaction
from core.fastapi.auth import hash_password, check_password, AuthInfo, encode_jwt
class UserService:
@staticmethod
async def find(
id: int | None = None,
email: str | None = None,
nickname: str | None = None,
) -> User | None:
if id is None and email is None and nickname is None:
return None
id_cri = (User.id == id) if id is not None else sqlalchemy.true()
email_cri = (User.email == email) if email is not None else sqlalchemy.true()
nickname_cri = (User.nickname == nickname) if nickname is not None else sqlalchemy.true()
query = session.query(User).filter(id_cri & email_cri & nickname_cri)
return query.first()
@staticmethod
async def create(email: str, password: str, password2: str, nickname: str) -> User:
with Transaction():
if password != password2:
raise PasswordDoesNotMatchException
if await UserService.find(email=email, nickname=nickname) is not None:
raise DuplicatedUserInfoException
user = User(
email=email,
password=hash_password(password),
nickname=nickname
)
session.add(user)
return user
@staticmethod
async def login(email: str, password: str) -> Tuple[User, str]:
user = await UserService.find(email=email)
if user is None:
raise UserNotFoundException
if not check_password(password, user.password):
raise IncorrectPasswordException
auth = AuthInfo(id=user.id)
jwt = f'bearer {encode_jwt(auth)}'
return user, jwt
__all__ = [
'UserService'
]
\ No newline at end of file
from pydantic import BaseModel
class MemoRequestView(BaseModel):
title: str
content: str
class PatchMemoRequestView(BaseModel):
title: str | None = None
content: str | None = None
class MemoResponseView(BaseModel):
class Writer(BaseModel):
id: int
nickname: str
class Config:
orm_mode = True
id: int
title: str
content: str
writer: Writer
class Config:
orm_mode = True
__all__ = [
'MemoRequestView',
'PatchMemoRequestView',
'MemoResponseView'
]
\ No newline at end of file
from pydantic import BaseModel
class CreateUserRequestView(BaseModel):
email: str
password: str
password2: str
nickname: str
class CreateUserResponseView(BaseModel):
id: int
email: str
nickname: str
class Config:
orm_mode = True
class GetUserResponseView(BaseModel):
id: int
email: str
nickname: str
class Config:
orm_mode = True
class LoginRequestView(BaseModel):
email: str
password: str
class LoginResponseView(BaseModel):
email: str
nickname: str
token: str
__all__ = [
'CreateUserRequestView',
'CreateUserResponseView',
'GetUserResponseView',
'LoginRequestView',
'LoginResponseView',
]
\ No newline at end of file
import os
from pydantic import BaseSettings
from dotenv import load_dotenv
load_dotenv()
class Config(BaseSettings):
APP_TITLE = 'boilerplate'
APP_DESCRIPTION = 'fastapi boilerplate project'
APP_VERSION = '1.0.0'
ENV: str = os.environ.get('ENV', 'dev')
DEBUG: bool = os.environ.get('DEBUG', True)
APP_HOST: str = os.environ.get('APP_HOST', '0.0.0.0')
APP_PORT: int = os.environ.get('APP_PORT', 5000)
DB_HOST: str | None = os.environ.get('DB_HOST')
DB_PORT: str | None = os.environ.get('DB_PORT')
DB_AUTHEN: str | None = os.environ.get('DB_AUTHEN')
DB_NAME: str | None = os.environ.get('DB_NAME')
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'SECRET_KEY')
JWT_ALGORITHM = 'HS256'
config: Config = Config()
__all__ = [
'config'
]
\ No newline at end of file
from .session import session
from .engine import get_database_url, disconnect
from .transaction import Transaction
from .mixins import TimestampMixin
from sqlalchemy.ext.declarative import declarative_base, declared_attr
import inflect
inflect_engine = inflect.engine()
class _ModelBase(object):
@declared_attr
def __tablename__(cls):
return inflect_engine.plural(cls.__name__.lower())