第2回 AWS 上的 Twelve-Factor 应用与 Django（创建后端 API）
目录
-
- 第1回 The Twelve-Factor App on AWS & Django（The Twelve-Factor Appとは）
- 第2回 The Twelve-Factor App on AWS & Django（バックエンドAPIを作ろう） ← 今回
首先
上次（第一篇）我们讨论了使用Twelve-Factor App进行开发的目标和意义。
我希望从本次开始，按照下图所示进行实施并推进开发。由于建立基础设施会涉及费用，所以我想先从应用的实施开始。
创建应用程序
我想要创建一个管理所有人都曾使用过的任务的TODO应用程序。
TODO应用程序的后端将使用Django作为RESTful API，前端将使用Nuxt作为SPA创建。
我想要创建后端API。我将使用以下库和版本。
言語/FW/ライブラリバージョンmacOS12.3.1Python3.10.6pyenv2.3.0-49-gae22c695pip22.2.2pipenv2022.8.5Django4.1djangorestframework3.13.1pytest7.1.2pytest-django4.5.2pytest-factoryboy2.5.0pytest-freezegun0.4.2pytest-cov3.0.0
首先，我们将进行Python环境的设置。为了确保后续能够切换Python版本，我们将使用pyenv进行安装。请根据您的环境进行安装。以下是在Mac上进行的安装步骤。
% brew update
% brew install pyenv
% echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.zshrc
% echo 'command -v pyenv >/dev/null || export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.zshrc
% echo 'eval "$(pyenv init -)"' >> ~/.zshrc
% exec "$SHELL"
一旦成功安装了pyenv，就可以开始创建应用程序了。
// アプリを格納するディレクトリを作成
% mkdir sample-ecs-todo-app
// 作成したディレクトリに移動
% cd sample-ecs-todo-app
// Python3.10.6をインストール
% pyenv install 3.10.6
// 使用するPythonのバージョンを3.10.6に設定
% pyenv global 3.10.6
// 作成したPython環境のpipを最新のバージョンにアップデート
% pip install --upgrade pip
// 作成したPython環境にpipenvをインストール
% pip install pipenv
// Djangoプロジェクトを作成するため、作成したPython環境にインストール
% pip install django
// バックエンドのアプリを格納するディレクトリを作成
% mkdir backend
// 作成したディレクトリ配下に移動
% cd backend
// django-adminコマンドでプロジェクトを作成
% django-admin startproject config .
// pipenvの仮想環境を
% export PIPENV_VENV_IN_PROJECT=true
// Python3.10.6で仮想環境を作成
% pipenv --python 3.10.6
// Djangoをインストール
% pipenv install django
// 環境変数をDjangoで扱いやすくするライブラリをインストール
% pipenv install django-environ
// PostgreSQLへアクセスするためのライブラリをインストール
% pipenv install psycopg2-binary
// Gunicorn（Python HTTP サーバー）ライブラリをインストール
% pipenv install gunicorn
// マイグレーションを実行
% python manage.py migrate
// マイグレーションを確認
% python manage.py showmigrations
// スーパーユーザーを作成
% python manage.py createsuperuser
Username (leave blank to use 'admin'): admin
Email address: admin@example.com
Password:
Password (again):
Superuser created successfully.
// 開発サーバーを起動
% python manage.py runserver
// DjangoでRESTful APIを作成するため、djangorestframeworkをインストール
% pipenv install djangorestframework
// TODOアプリを作成
% python manage.py startapp todo
当您创建了TODO应用程序的模板后，您将创建并编辑以下文件。
from pathlib import Path
import environ
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/4.1/howto/deployment/checklist/
+ env = environ.Env()
+ dot_env_path = BASE_DIR / '.env'
+ if dot_env_path.exists():
+ env.read_env(str(dot_env_path))
# SECURITY WARNING: keep the secret key used in production secret!
- SECRET_KEY = 'django-insecure-5&cyjjqq=+l^mye=6ton2m%f6=)3vs)q%7^4^66sl_y+y_+tdw'
+ SECRET_KEY = env.str('SECRET_KEY', 'django-insecure-5&cyjjqq=+l^mye=6ton2m%f6=)3vs)q%7^4^66sl_y+y_+tdw')
# SECURITY WARNING: don't run with debug turned on in production!
- DEBUG = True
+ DEBUG = env.bool('DEBUG', False)
- ALLOWED_HOSTS = []
+ ALLOWED_HOSTS = env.list('ALLOWED_HOSTS', default=['127.0.0.1', 'localhost'])
+ LOCAL_DEV = env.bool('LOCAL_DEV', default=False)
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
+ 'rest_framework',
+ 'todo',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
'middlewares.RequestLogMiddleware',
]
ROOT_URLCONF = 'config.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'config.wsgi.application'
# Database
# https://docs.djangoproject.com/en/4.1/ref/settings/#databases
+ if LOCAL_DEV:
+ DATABASES = {
+ 'default': {
+ 'ENGINE': 'django.db.backends.sqlite3',
+ 'NAME': BASE_DIR / 'db.sqlite3',
+ 'TEST': {
+ 'CHARSET': 'UTF8',
+ 'NAME': ':memory:',
+ },
+ }
+ }
+ else:
+ DATABASES = {
+ 'default': env.db(),
+ }
# Password validation
# https://docs.djangoproject.com/en/4.1/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
# Internationalization
# https://docs.djangoproject.com/en/4.1/topics/i18n/
- LANGUAGE_CODE = 'en-us'
+ LANGUAGE_CODE = 'ja'
- TIME_ZONE = 'UTC'
+ TIME_ZONE = 'Asia/Tokyo'
USE_I18N = True
USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/4.1/howto/static-files/
STATIC_URL = 'static/'
# Default primary key field type
# https://docs.djangoproject.com/en/4.1/ref/settings/#default-auto-field
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
+ # logging
+ LOGGING = {
+ 'version': 1,
+ 'disable_existing_loggers': True,
+ 'formatters': {
+ 'backend': {
+ '()': 'formatter.JsonFormatter',
+ 'format': '[%(levelname)s]\t%(asctime)s.%(msecs)dZ\t%(levelno)s\t%(message)s\n'
+ }
+ },
+ 'handlers': {
+ 'console': {
+ 'level': 'INFO',
+ 'class': 'logging.StreamHandler',
+ 'formatter': 'backend',
+ },
+ },
+ 'loggers': {
+ 'django.request': {
+ 'handlers': ['console'],
+ 'level': 'INFO',
+ },
+ 'django.db.backends': {
+ 'handlers': ['console'],
+ 'level': 'DEBUG',
+ },
+ 'backend': {
+ 'handlers': ['console'],
+ 'level': 'DEBUG',
+ 'propagate': False,
+ },
+ },
+ }
为了在本地环境中开发，将环境变量定义在.env文件中。
+ SECRET_KEY=django-insecure-5&cyjjqq=+l^mye=6ton2m%f6=)3vs)q%7^4^66sl_y+y_+tdw
+ DEBUG=True
+ LOCAL_DEV=True
为了方便在AWS CloudWatch日志中查看日志，我们将创建一个格式化程序来将日志转换为JSON格式。
+ import json
+ import logging
+ import traceback
+
+
+ class JsonFormatter(logging.Formatter):
+ def format(self, record):
+ if self.usesTime():
+ record.asctime = self.formatTime(record, self.datefmt)
+
+ json_log = {
+ 'aws_request_id': getattr(record, 'aws_request_id', '00000000-0000-0000-0000-000000000000'),
+ 'log_level': str(getattr(record, 'levelname', '')),
+ 'timestamp': '%(asctime)s.%(msecs)dZ' % dict(asctime=record.asctime, msecs=record.msecs),
+ 'aws_request_id': getattr(record, 'aws_request_id', '00000000-0000-0000-0000-000000000000'),
+ 'message': record.getMessage(),
+ 'status_code': str(getattr(record, 'status_code', '')),
+ 'execution_time': str(getattr(record, 'execution_time', '')),
+ 'stack_trace': {},
+ }
+
+ request = getattr(record, 'request', None)
+
+ if request:
+ json_log = {
+ 'aws_cf_id': request.META.get('HTTP_X_AMZ_CF_ID', ''),
+ 'aws_trace_id': request.META.get('HTTP_X_AMZN_TRACE_ID', ''),
+ 'x_real_ip': request.META.get('HTTP_X_REAL_IP', ''),
+ 'x_forwarded_for': request.META.get('HTTP_X_FORWARDED_FOR', ''),
+ 'request_method': request.method,
+ 'request_path': request.get_full_path(),
+ 'request_body': request.request_body,
+ 'user_agent': request.META.get('HTTP_USER_AGENT', ''),
+ 'user': str(getattr(request, 'user', '')),
+ 'stack_trace': {},
+ }
+
+ if record.exc_info:
+ json_log['stack_trace'] = traceback.format_exc().splitlines()
+
+ return json.dumps(json_log, ensure_ascii=False)
我们将创建一个使用格式化程序的中间件，以便能够以JSON格式输出HTTP请求的日志。
+ import copy
+ import json
+ import logging
+ import re
+ import time
+
+ from django.utils.deprecation import MiddlewareMixin
+
+ request_logger = logging.getLogger('backend')
+
+
+ class RequestLogMiddleware(MiddlewareMixin):
+ """Request Logging Middleware."""
+
+ def __init__(self, *args, **kwargs):
+ """Constructor method."""
+ super().__init__(*args, **kwargs)
+
+ def is_json_format(self, request_body):
+ try:
+ json.loads(request_body)
+ except json.JSONDecodeError:
+ return False
+
+ return True
+
+ def process_request(self, request):
+ """Set Request Start Time to measure time taken to service request."""
+ request.start_time = time.time()
+ request.request_body = ''
+ if request.method in ['PUT', 'POST', 'PATCH']:
+ try:
+ request_body = request.body.decode('utf-8')
+
+ if request_body and self.is_json_format(request_body):
+ request.request_body = json.loads(request_body)
+ else:
+ request.request_body = request_body
+ except UnicodeDecodeError:
+ # request.request_body = str(base64.b64encode(request.body))
+ request_body = copy.deepcopy(request.body)
+ first_file_name = ''
+ file_names = re.findall('filename=\".*\"\\\\r\\\\nContent-Type', str(request_body))
+ if len(file_names) > 0:
+ first_file_name = file_names[0]
+ first_file_name = first_file_name.replace('\\r\\nContent-Type', '')
+ first_file_name = first_file_name.replace('\\\\', '\\')
+ request.request_body = first_file_name
+
+ def process_response(self, request, response):
+ status_code = getattr(response, 'status_code', '')
+ log_info = {
+ 'request': request,
+ 'status_code': status_code,
+ 'execution_time': time.time() - request.start_time,
+ }
+ if status_code == 200:
+ request_logger.info(msg='OK', extra=log_info)
+ return response
我会创建一个模型，将Todo存储到数据库中。
+ from django.db import models
+
+
+ class Todo(models.Model):
+ ACTIVE = 0
+ DONE = 1
+ STATUS = ((ACTIVE, 'active'), (DONE, 'done'))
+
+ title = models.CharField(max_length=30)
+ description = models.TextField(blank=True, null=True)
+ status = models.SmallIntegerField(choices=STATUS, default=ACTIVE)
+ created_at = models.DateTimeField(auto_now_add=True)
我們將通過Django的管理界面來修改Todo表的數據。
+ from django.contrib import admin
+
+ from todo import models
+
+
+ @admin.register(models.Todo)
+ class TodoAdmin(admin.ModelAdmin):
+ list_display = ['id', 'title', 'description', 'status', 'created_at']
+ search_fields = ['title']
定义从数据库获取的Todo数据的输入和输出。
+ from rest_framework import serializers
+
+ from todo.models import Todo
+
+
+ class TodoSerializer(serializers.ModelSerializer):
+ class Meta:
+ model = Todo
+ fields = ['id', 'title', 'description', 'created_at']
+
+ def create(self, validated_data):
+ return Todo.objects.create(**validated_data)
+
+ def update(self, instance, validated_data):
+ instance.status = validated_data.get('status', instance.status)
+ instance.save()
+ return instance
我们将定义一个用于获取和发送Todo的API。
+ from rest_framework import viewsets
+
+ from todo.models import Todo
+ from todo.serializers import TodoSerializer
+
+
+ class TodoViewSet(viewsets.ModelViewSet):
+ queryset = Todo.objects.all().order_by('-id')
+ serializer_class = TodoSerializer
将已创建的Todo API与URL关联起来。
+ from rest_framework import routers
+
+ from todo.views import TodoViewSet
+
+ router = routers.DefaultRouter()
+ router.register(r'todos', TodoViewSet)
将定义的URL公开给外部。
+ from todo.urls import router
urlpatterns = [
path('admin/', admin.site.urls),
+ path('api/', include(router.urls)),
]
完成上述代码的附加和编辑后，请执行以下操作以进行数据库迁移。
// マイグレーションファイルを作成
% python manage.py makemigrations
// マイグレーションを確認
% python manage.py showmigrations
// マイグレーションを実行
% python manage.py migrate
请访问网址 http://127.0.0.1:8000/admin/，进入Django的管理界面，并使用上述创建的超级用户账户登录。
如果您成功登录，请点击“添加”按钮进入添加页面，输入相关信息并点击“保存”按钮来创建待办事项。
如果成功创建，则可以访问http://127.0.0.1:8000/api/todos/，就可以以JSON格式获取创建的TODO，如下图所示。
虽然简单，但我成功创建了后端的 API。
我希望在后续的工作中将自动测试应用到之前创建的API中，并在将其作为CI推送到Github后自动执行测试，并输出测试覆盖率。
首先安装实现自动化测试所需的库。
// pytest（Pythonでテストを実装するための使い勝手が良いライブラリ）をインストール
% pipenv install --dev pytest
// pytest-django（Djangoでpytestを扱いやすくするライブラリ）をインストール
% pipenv install --dev pytest-django
// pytest-factoryboy（テストデータを作成しやすくするライブラリ）をインストール
% pipenv install --dev pytest-factoryboy
// pytest-freezegun（時刻を固定してテストしやすくするライブラリ）をインストール
% pipenv install --dev pytest-freezegun
// pytest-cov（カバレッジを取得するライブラリ）をインストール
% pipenv install --dev pytest-cov
如果能够成功安装库文件，那么就创建以下的文件。
+
我将定义一个使得创建测试数据更容易的类。
+ import factory
+
+ from todo.models import Todo
+
+
+ class TodoFactory(factory.django.DjangoModelFactory):
+ class Meta:
+ model = Todo
+
+ id = 1
+ title = 'Implement API'
+ description = 'Implement an API to retrieve Todo'
+ status = 0
我将实施API的自动化测试。我会注册两个ToDo，并通过API确认可以按照注册顺序（逆序）获取ToDo。
+ import datetime
+ import json
+ import zoneinfo
+
+ import pytest
+ from rest_framework.test import APIRequestFactory
+
+ from todo.tests.factories import TodoFactory
+ from todo.views import TodoViewSet
+
+
+ @pytest.mark.freeze_time(datetime.datetime(2022, 8, 11, 9, 0, 0, tzinfo=zoneinfo.ZoneInfo('Asia/Tokyo')))
+ @pytest.mark.django_db
+ def tests_should_get_two_todos():
+ TodoFactory()
+ TodoFactory(id=2, title='Code Review', description='Review Pull Request #1')
+
+ client = APIRequestFactory()
+ todo_list = TodoViewSet.as_view({'get': 'list'})
+
+ request = client.get('/api/todos/')
+ response = todo_list(request)
+ response.render()
+
+ assert response.status_code == 200
+ assert json.loads(response.content) == [
+ {
+ 'id': 2,
+ 'title': 'Code Review',
+ 'description': 'Review Pull Request #1',
+ 'status': 0,
+ 'created_at': '2022-08-11T09:00:00+09:00'
+ },
+ {
+ 'id': 1,
+ 'title': 'Implement API',
+ 'description': 'Implement an API to retrieve Todo',
+ 'status': 1,
+ 'created_at': '2022-08-11T09:00:00+09:00'
+ }
+ ]
定义pytest的设置。
+ [pytest]
+ DJANGO_SETTINGS_MODULE = config.settings
+ python_files = tests_*.py
如果可以创建上述文件，请执行pytest –cov –cov-report=html -v。如果可以执行，应该可以输出下面的测试结果。
% pytest --cov --cov-report=html -v
======================================================================================= test session starts ========================================================================================
platform darwin -- Python 3.10.6, pytest-7.1.2, pluggy-1.0.0 -- /Users/staff/Dev/sample-ecs-todo-app/backend/.venv/bin/python
cachedir: .pytest_cache
django: settings: config.settings (from ini)
rootdir: /Users/staff/Dev/sample-ecs-todo-app/backend, configfile: pytest.ini
plugins: freezegun-0.4.2, factoryboy-2.5.0, Faker-13.15.1, django-4.5.2, cov-3.0.0
collected 1 item
todo/tests/tests_views.py::tests_should_get_two_todos PASSED
我成功通过了自动化测试，并确认API按照预期运行，一切正常。
覆盖率也会输出到backend/htmlcov目录中，打开index.html即可查看每个文件的覆盖率。点击文件可以查看该文件中已覆盖和未覆盖的代码。
打开backend/todo/serializers.py文件时，它会以绿色突出显示已覆盖的代码，以红色突出显示未覆盖的代码。
以上是后端实现的完成。最终的目录结构如下所示。
sample-ecs-todo-app
└── backend
├── Pipfile
├── Pipfile.lock
├── README.md
├── __init__.py
├── config
│ ├── __init__.py
│ ├── asgi.py
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
├── db.sqlite3
├── manage.py
├── pytest.ini
└── todo
├── __init__.py
├── admin.py
├── apps.py
├── migrations
│ ├── 0001_initial.py
│ └── __init__.py
├── models.py
├── serializers.py
├── tests
│ ├── __init__.py
│ ├── factories.py
│ └── tests_views.py
├── tests.py
├── urls.py
└── views.py
请试着克隆一下上述代码，因为我已经将其推送到Github上了。
下一次我想要用Nuxt来创建前端的SPA。
