Flask flask-restx mvc패턴

# 프로젝트 구조
my_flask_app/
│
├── app/
│   ├── __init__.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── task.py
│   ├── views/
│   │   ├── __init__.py
│   │   └── task_view.py
│   ├── controllers/
│   │   ├── __init__.py
│   │   └── task_controller.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── task_schema.py
│   └── utils/
│       ├── __init__.py
│       └── logger.py
│
├── config.py
├── tests/
│   ├── __init__.py
│   ├── test_models.py
│   ├── test_views.py
│   └── test_controllers.py
│
├── .env
├── .gitignore
├── requirements.txt
├── run.py
└── wsgi.py

# .env
FLASK_APP=run.py
FLASK_ENV=development
SECRET_KEY=your_secret_key_here
DATABASE_URL=sqlite:///dev.db
HOST=0.0.0.0
PORT=5000

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    HOST = os.environ.get('HOST', '0.0.0.0')
    PORT = int(os.environ.get('PORT', 5000))

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///dev.db'

class ProductionConfig(Config):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///test.db'

config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'testing': TestingConfig,
    'default': DevelopmentConfig
}

# requirements.txt
Flask==2.0.1
flask-restx==0.5.1
Flask-SQLAlchemy==2.5.1
marshmallow==3.13.0
python-dotenv==0.19.0
gunicorn==20.1.0
pytest==6.2.5

# app/__init__.py
import os
from flask import Flask
from flask_restx import Api
from flask_sqlalchemy import SQLAlchemy
from .utils.logger import setup_logger
from .config import config

# 데이터베이스 인스턴스 생성
db = SQLAlchemy()

def create_app(config_name=None):
    """애플리케이션 팩토리 함수"""
    app = Flask(__name__)

    # 환경 변수에서 설정 가져오기 (기본값은 'development')
    config_name = config_name or os.getenv('FLASK_ENV', 'development')
    
    # 설정 로드
    app.config.from_object(config[config_name])
    
    # 데이터베이스 초기화
    db.init_app(app)

    # API 초기화
    api = Api(app, version='1.0', title='Task API', description='A simple Task API')

    # 로거 설정
    setup_logger(app)

    # 뷰 등록
    from .views.task_view import api as task_ns
    api.add_namespace(task_ns)

    # 데이터베이스 생성
    with app.app_context():
        db.create_all()

    return app

# app/models/task.py
from app import db

class Task(db.Model):
    """작업 모델"""
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    description = db.Column(db.String(200))
    done = db.Column(db.Boolean, default=False)

    def __repr__(self):
        return f'<Task {self.id}: {self.title}>'

# app/schemas/task_schema.py
from marshmallow import Schema, fields

class TaskSchema(Schema):
    """작업 스키마"""
    id = fields.Int(dump_only=True)
    title = fields.Str(required=True)
    description = fields.Str()
    done = fields.Boolean()

task_schema = TaskSchema()
tasks_schema = TaskSchema(many=True)

# app/controllers/task_controller.py
from app.models.task import Task
from app.schemas.task_schema import task_schema, tasks_schema
from app import db

class TaskController:
    """작업 컨트롤러"""

    @staticmethod
    def get_all_tasks():
        """모든 작업을 조회"""
        tasks = Task.query.all()
        return tasks_schema.dump(tasks)

    @staticmethod
    def get_task(task_id):
        """특정 작업을 조회"""
        task = Task.query.get(task_id)
        return task_schema.dump(task) if task else None

    @staticmethod
    def create_task(data):
        """새 작업을 생성"""
        task = Task(**task_schema.load(data))
        db.session.add(task)
        db.session.commit()
        return task_schema.dump(task)

    @staticmethod
    def update_task(task_id, data):
        """작업을 수정"""
        task = Task.query.get(task_id)
        if task:
            task.title = data.get('title', task.title)
            task.description = data.get('description', task.description)
            task.done = data.get('done', task.done)
            db.session.commit()
            return task_schema.dump(task)
        return None

    @staticmethod
    def delete_task(task_id):
        """작업을 삭제"""
        task = Task.query.get(task_id)
        if task:
            db.session.delete(task)
            db.session.commit()
            return True
        return False

# app/views/task_view.py
from flask_restx import Namespace, Resource, fields
from app.controllers.task_controller import TaskController
from flask import current_app as app

api = Namespace('tasks', description='Task operations')

# API 모델 정의
task_model = api.model('Task', {
    'id': fields.Integer(readonly=True, description='The task unique identifier'),
    'title': fields.String(required=True, description='The task title'),
    'description': fields.String(description='The task description'),
    'done': fields.Boolean(description='The task status')
})

@api.route('/')
class TaskList(Resource):
    @api.doc('list_tasks')
    @api.marshal_list_with(task_model)
    def get(self):
        """모든 작업을 조회"""
        app.logger.info('Fetching all tasks')
        return TaskController.get_all_tasks()

    @api.doc('create_task')
    @api.expect(task_model)
    @api.marshal_with(task_model, code=201)
    def post(self):
        """새 작업을 생성"""
        app.logger.info('Creating a new task')
        return TaskController.create_task(api.payload), 201

@api.route('/<int:id>')
@api.param('id', 'The task identifier')
@api.response(404, 'Task not found')
class Task(Resource):
    @api.doc('get_task')
    @api.marshal_with(task_model)
    def get(self, id):
        """특정 작업을 조회"""
        app.logger.info(f'Fetching task with id: {id}')
        task = TaskController.get_task(id)
        return task if task else api.abort(404, f"Task {id} doesn't exist")

    @api.doc('update_task')
    @api.expect(task_model)
    @api.marshal_with(task_model)
    def put(self, id):
        """작업을 수정"""
        app.logger.info(f'Updating task with id: {id}')
        task = TaskController.update_task(id, api.payload)
        return task if task else api.abort(404, f"Task {id} doesn't exist")

    @api.doc('delete_task')
    @api.response(204, 'Task deleted')
    def delete(self, id):
        """작업을 삭제"""
        app.logger.info(f'Deleting task with id: {id}')
        if TaskController.delete_task(id):
            return '', 204
        api.abort(404, f"Task {id} doesn't exist")

# app/utils/logger.py
import logging
from flask import request

def setup_logger(app):
    """로거 설정"""
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s '
        '[in %(pathname)s:%(lineno)d]'
    ))
    app.logger.addHandler(handler)
    app.logger.setLevel(logging.INFO)

    @app.before_request
    def log_request_info():
        app.logger.info('Headers: %s', request.headers)
        app.logger.info('Body: %s', request.get_data())

# run.py
from app import create_app

app = create_app()

if __name__ == '__main__':
    app.run(host=app.config['HOST'], port=app.config['PORT'])

# wsgi.py
from app import create_app

application = create_app('production')

if __name__ == "__main__":
    application.run(host=application.config['HOST'], port=application.config['PORT'])

# tests/test_models.py
import pytest
from app.models.task import Task

def test_new_task():
    """
    GIVEN a Task model
    WHEN a new Task is created
    THEN check the title, description, and done fields are defined correctly
    """
    task = Task(title='New task', description='Test description')
    assert task.title == 'New task'
    assert task.description == 'Test description'
    assert task.done == False

# tests/test_views.py
import json
import pytest
from app import create_app, db

@pytest.fixture
def client():
    app = create_app('testing')
    app.config['TESTING'] = True
    
    with app.test_client() as client:
        with app.app_context():
            db.create_all()
        yield client
        with app.app_context():
            db.drop_all()

def test_get_tasks(client):
    """
    GIVEN a Flask application
    WHEN the '/tasks' page is requested (GET)
    THEN check the response is valid
    """
    response = client.get('/tasks/')
    assert response.status_code == 200
    assert b'[]' in response.data

def test_create_task(client):
    """
    GIVEN a Flask application
    WHEN a new task is created (POST)
    THEN check the response is valid and the task is created
    """
    response = client.post('/tasks/', json={'title': 'Test task', 'description': 'Test description'})
    assert response.status_code == 201
    data = json.loads(response.data)
    assert data['title'] == 'Test task'
    assert data['description'] == 'Test description'

# tests/test_controllers.py
import pytest
from app import create_app, db
from app.controllers.task_controller import TaskController
from app.models.task import Task

@pytest.fixture
def app():
    app = create_app('testing')
    app.config['TESTING'] = True
    return app

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def init_database(app):
    with app.app_context():
        db.create_all()
        yield db
        db.drop_all()

def test_create_task(init_database):
    """
    GIVEN a Task controller
    WHEN a new Task is created
    THEN check the title, description, and done fields are defined correctly
    """
    task_data = {'title': 'New task', 'description': 'Test description'}
    task = TaskController.create_task(task_data)
    assert task['title'] == 'New task'
    assert task['description'] == 'Test description'
    assert task['done'] == False

def test_get_all_tasks(init_database):
    """
    GIVEN a Task controller
    WHEN all tasks are requested
    THEN check if the correct number of tasks is returned
    """
    TaskController.create_task({'title': 'Task 1'})
    TaskController.create_task({'title': 'Task 2'})
    tasks = TaskController.get_all_tasks()
    assert len(tasks) == 2

개발 모드 실행: FLASK_ENV=development python run.py
프로덕션 모드 실행: FLASK_ENV=production gunicorn --workers 4 --worker-class gthread --threads 2 -b 0.0.0.0:5000 wsgi:app
테스트 모드 실행: FLASK_ENV=testing pytest