курсовая_v1. Курсовая работа Необходимо
 Скачать 2.93 Mb.
|
|
Настраиваем Alembic Когда схема базы данных описана, необходимо сгенерировать миграции, но для этого сначала нужно настроить Alembic. Чтобы воспользоваться командой alembic, необходимо выполнить следующие шаги: 1. Установить пакет: pip install alembic 2. Инициализировать Alembic: cd analyzer && alembic init db/alembic. Эта команда создаст файл конфигурации analyzer/alembic.ini и папку analyzer/db/alembic со следующим содержимым: • env.py — вызывается каждый раз при запуске Alembic. Подключает в Alembic реестр sqlalchemy.MetaData с описанием желаемого состояния БД и содержит инструкции по запуску миграций. • script.py.mako — шаблон, на основе которого генерируются миграции. • versions — папка, в которой Alembic будет искать (и генерировать) миграции. 3. Указать адрес базы данных в файле alembic.ini: ; analyzer/alembic.ini [alembic] sqlalchemy.url = postgresql://user:hackme@localhost/analyzer 4. Указать описание желаемого состояния базы данных (реестр sqlalchemy.MetaData), чтобы Alembic мог генерировать миграции автоматически: # analyzer/db/alembic/env.py from analyzer.db import schema target_metadata = schema.metadata Alembic настроен и им уже можно пользоваться, но в нашем случае такая конфигурация имеет ряд недостатков: 1. Утилита alembic ищет alembic.ini в текущей рабочей директории. Путь к alembic.ini можно указать аргументом командной строки, но это неудобно: хочется иметь возможность вызывать команду из любой папки без дополнительных параметров. 2. Чтобы настроить Alembic на работу с определенной базой данных, требуется менять файл alembic.ini . Гораздо удобнее было бы указать настройки БД переменной окружения и/или аргументом командной строки, например --pg-url 3. Название утилиты alembic не очень хорошо коррелирует с названием нашего сервиса (а пользователь фактически может вообще не владеть Python и ничего не знать об Alembic). Конечному пользователю было бы намного удобнее, если бы все исполняемые команды сервиса имели общий префикс, например analyzer-* Эти проблемы решаются с помощью небольшой обертки analyzer/db/__main__.py: • Для обработки аргументов командной строки Alembic использует стандартный модуль argparse . Он позволяет добавить необязательный аргумент --pg-url со значением по умолчанию из переменной окружения ANALYZER_PG_URL Код import os from alembic.config import CommandLine, Config from analyzer.utils.pg import DEFAULT_PG_URL def main () : alembic = CommandLine() alembic.parser.add_argument( '--pg-url' , default=os.getenv( 'ANALYZER_PG_URL' , DEFAULT_PG_URL), help= 'Database URL [env var: ANALYZER_PG_URL]' ) options = alembic.parser.parse_args() # Создаем объект конфигурации Alembic config = Config(file_=options.config, ini_section=options.name, cmd_opts=options) # Меняем значение sqlalchemy.url из конфига Alembic config.set_main_option( 'sqlalchemy.url' , options.pg_url) # Запускаем команду alembic exit(alembic.run_cmd(config, options)) if __name__ == '__main__' : main() • Путь до файла alembic.ini можно рассчитывать относительно расположения исполняемого файла, а не текущей рабочей директории пользователя. Код import os from alembic.config import CommandLine, Config from pathlib import Path PROJECT_PATH = Path(__file__).parent.parent.resolve() def main () : alembic = CommandLine() options = alembic.parser.parse_args() # Если указан относительный путь (alembic.ini), добавляем в начало # абсолютный путь до приложения if not os.path.isabs(options.config): options.config = os.path.join(PROJECT_PATH, options.config) # Создаем объект конфигурации Alembic config = Config(file_=options.config, ini_section=options.name, cmd_opts=options) # Подменяем путь до папки с alembic на абсолютный (требуется, чтобы alembic # мог найти env.py, шаблон для генерации миграций и сами миграции) alembic_location = config.get_main_option( 'script_location' ) if not os.path.isabs(alembic_location): config.set_main_option( 'script_location' , os.path.join(PROJECT_PATH, alembic_location)) # Запускаем команду alembic exit(alembic.run_cmd(config, options)) if __name__ == '__main__' : main() Когда утилита для управления состоянием БД готова, ее можно зарегистрировать в setup.py как исполняемую команду с понятным конечному пользователю названием, например analyzer-db: Регистрация исполняемой команды в setup.py from setuptools import setup setup(..., entry_points={ 'console_scripts' : [ 'analyzer-db = analyzer.db.__main__:main' ] }) После переустановки модуля будет сгенерирован файл env/bin/analyzer-db и команда analyzer-db станет доступной: $ pip install -e '.[dev]' """ Утилита для управления состоянием базы данных, обертка над alembic. Можно вызывать из любой директории, а также указать произвольный DSN для базы данных, отличный от указанного в файле alembic.ini. """ import argparse import logging import os from alembic.config import CommandLine from analyzer.utils.pg import DEFAULT_PG_URL, make_alembic_config def main(): logging.basicConfig(level=logging.DEBUG) alembic = CommandLine() alembic.parser.formatter_class = argparse.ArgumentDefaultsHelpFormatter alembic.parser.add_argument( '--pg-url', default=os.getenv('ANALYZER_PG_URL', DEFAULT_PG_URL), help='Database URL [env var: ANALYZER_PG_URL]' ) options = alembic.parser.parse_args() if 'cmd' not in options: alembic.parser.error('too few arguments') exit(128) else: config = make_alembic_config(options) exit(alembic.run_cmd(config, options)) if __name__ == '__main__': main() Генерируем миграции Чтобы сгенерировать миграции, требуется два состояния: желаемое (которое мы описали объектами SQLAlchemy) и реальное (база данных, в нашем случае пустая). Я решил, что проще всего поднять Postgres с помощью Docker и для удобства добавил команду make postgres , запускающую в фоновом режиме контейнер с PostgreSQL на 5432 порту: Поднимаем PostgreSQL и генерируем миграцию $ make postgres $ analyzer-db revision --message= "Initial" --autogenerate INFO [alembic.runtime.migration] Context impl PostgresqlImpl. INFO [alembic.runtime.migration] Will assume transactional DDL. INFO [alembic.autogenerate.compare] Detected added table 'imports' INFO [alembic.autogenerate.compare] Detected added table 'citizens' INFO [alembic.autogenerate.compare] Detected added index 'ix__citizens__town' on '[' town ']' INFO [alembic.autogenerate.compare] Detected added table 'relations' Generating /Users/alvassin/Work/backendschool2019/analyzer/db/alembic/versions/d5f704ed4 610_initial.py ... done Alembic в целом хорошо справляется с рутинной работой генерации миграций, но я хотел бы обратить внимание на следующее: • Пользовательские типы данных, указанные в создаваемых таблицах, создаются автоматически (в нашем случае — gender ), но код для их удаления в downgrade не генерируется. Если применить, откатить и потом еще раз применить миграцию, это вызовет ошибку, так как указанный тип данных уже существует. Удаляем тип данных gender в методе downgrade from alembic import op from sqlalchemy import Column, Enum GenderType = Enum( 'female' , 'male' , name= 'gender' ) def upgrade () : # При создании таблицы тип данных GenderType будет создан автоматически op.create_table( 'citizens' , ..., Column( 'gender' , GenderType, nullable= False )) def downgrade () : op.drop_table( 'citizens' ) # После удаления таблицы тип данных необходимо удалить GenderType.drop(op.get_bind()) • В методе downgrade некоторые действия иногда можно убрать (если мы удаляем таблицу целиком, можно не удалять ее индексы отдельно): Например def downgrade () : op.drop_table( 'relations' ) # Следующим шагом мы удаляем таблицу citizens, индекс будет удален автоматически # эту строчку можно удалить op.drop_index(op.f( 'ix__citizens__town' ), table_name= 'citizens' ) op.drop_table( 'citizens' ) op.drop_table( 'imports' ) Когда миграция исправлена и готова, применим ее: $ analyzer-db upgrade head INFO [alembic.runtime.migration] Context impl PostgresqlImpl. INFO [alembic.runtime.migration] Will assume transactional DDL. INFO [alembic.runtime.migration] Running upgrade -> d5f704ed4610, Initial Приложение Прежде чем приступить к созданию обработчиков, необходимо сконфигурировать приложение aiohttp. Если посмотреть aiohttp quickstart, можно написать приблизительно такой код import logging from aiohttp import web def main () : # Настраиваем логирование logging.basicConfig(level=logging.DEBUG) # Создаем приложение app = web.Application() # Регистрируем обработчики app.router.add_route(...) # Запускаем приложение web.run_app(app) Этот код вызывает ряд вопросов и имеет ряд недостатков: • Как конфигурировать приложение? Как минимум, необходимо указать хост и порт для подключения клиентов, а также информацию для подключения к базе данных. Решать эту задачу можно с помощью модуля ConfigArgParse : он расширяет стандартный argparse и позволяет использовать для конфигурации аргументы командной строки, переменные окружения (незаменимые для конфигурации Docker-контейнеров) и даже файлы конфигурации (а также совмещать эти способы). C помощью ConfigArgParse также можно валидировать значения параметров конфигурации приложения. Пример обработки параметров с помощью ConfigArgParse from aiohttp import web from configargparse import ArgumentParser, ArgumentDefaultsHelpFormatter from analyzer.utils.argparse import positive_int parser = ArgumentParser( # Парсер будет искать переменные окружения с префиксом ANALYZER_, # например ANALYZER_API_ADDRESS и ANALYZER_API_PORT auto_env_var_prefix= 'ANALYZER_' , # Покажет значения параметров по умолчанию formatter_class=ArgumentDefaultsHelpFormatter ) parser.add_argument( '--api-address' , default= '0.0.0.0' , help= 'IPv4/IPv6 address API server would listen on' ) # Разрешает только целые числа больше нуля parser.add_argument( '--api-port' , type=positive_int, default= 8081 , help= 'TCP port API server would listen on' ) def main () : # Получаем параметры конфигурации, которые можно передать как аргументами # командной строки, так и переменными окружения args = parser.parse_args() # Запускаем приложение на указанном порту и адресе app = web.Application() web.run_app(app, host=args.api_address, port=args.api_port) if __name__ == '__main__' : main() Кстати, ConfigArgParse , как и argparse , умеет генерировать подсказку по запуску команды с описанием всех аргументов (необходимо позвать команду с аргументом -h или --help ). Это невероятно облегчает жизнь пользователям вашего ПО: • Например • $ python __main__.py -- help • usage: __main__.py [-h] [--api-address API_ADDRESS] [--api-port API_PORT] • • If an arg is specified in more than one place, then commandline values override environment variables which override defaults. • • optional arguments: • -h, -- help show this help message and exit • --api-address API_ADDRESS • IPv4/IPv6 address API server would listen on [env var: ANALYZER_API_ADDRESS] (default: 0.0.0.0) --api-port API_PORT TCP port API server would listen on [env var: ANALYZER_API_PORT] (default: 8081) • После получения переменные окружения больше не нужны и даже могут представлять опасность — например, они могут случайно «утечь» с отображением информации об ошибке. Злоумышленники в первую очередь будут пытаться получить информацию об окружении, поэтому очистка переменных окружения считается хорошим тоном. Можно было бы воспользоваться os.environ.clear() , но Python позволяет управлять поведением модулей стандартной библиотеки с помощью многочисленных переменных окружения (например, вдруг потребуется включить режим отладки asyncio ?), поэтому разумнее очищать переменные окружения по префиксу приложения, указанного в ConfigArgParser Пример import os from typing import Callable from configargparse import ArgumentParser from yarl import URL from analyzer.api.app import create_app from analyzer.utils.pg import DEFAULT_PG_URL ENV_VAR_PREFIX = 'ANALYZER_' parser = ArgumentParser(auto_env_var_prefix=ENV_VAR_PREFIX) parser.add_argument( '--pg-url' , type=URL, default=URL(DEFAULT_PG_URL), help= 'URL to use to connect to the database' ) def clear_environ (rule: Callable) : """ Очищает переменные окружения, переменные для очистки определяет переданная функция rule """ # Ключи из os.environ копируются в новый tuple, чтобы не менять объект # os.environ во время итерации for name in filter(rule, tuple(os.environ)): os.environ.pop(name) def main () : # Получаем аргументы args = parser.parse_args() # Очищаем переменные окружения по префиксу ANALYZER_ clear_environ( lambda i: i.startswith(ENV_VAR_PREFIX)) # Запускаем приложение app = create_app(args) if __name__ == '__main__' : main() • Запись логов в stderr/файл в основном потоке блокирует цикл событий. По умолчанию logging.basicConfig() настраивает запись логов в stderr Чтобы логирование не мешало эффективной работе асинхронного приложения, необходимо выполнять запись логов в отдельном потоке. Для этого можно воспользоваться готовым методом из модуля aiomisc. Настраиваем логирование с помощью aiomisc import logging from aiomisc.log import basic_config basic_config(logging.DEBUG, buffered= True ) • Как масштабировать приложение, если одного процесса станет недостаточно для обслуживания входящего трафика? Можно сначала аллоцировать сокет, затем с помощью fork создать несколько новых отдельных процессов, и соединения на сокете будут распределяться между ними механизмами ядра (конечно, под Windows это не работает). Пример import os from sys import argv import forklib from aiohttp.web import Application, run_app from aiomisc import bind_socket from setproctitle import setproctitle def main () : sock = bind_socket(address= '0.0.0.0' , port= 8081 , proto_name= 'http' ) setproctitle( f'[Master] {os.path.basename(argv[ 0 ])} ' ) def worker () : setproctitle( f'[Worker] {os.path.basename(argv[ 0 ])} ' ) app = Application() run_app(app, sock=sock) forklib.fork(os.cpu_count(), worker, auto_restart= True ) if __name__ == '__main__' : main() • Требуется ли приложению обращаться или аллоцировать какие-либо ресурсы во время работы? Если нет, по соображениям безопасности все ресурсы (в нашем случае — сокет для подключения клиентов) можно аллоцировать на старте, а затем сменить пользователя на nobody. Он обладает ограниченным набором привиллегий — это здорово усложнит жизнь злоумышленникам. Пример import os import pwd from aiohttp.web import run_app from aiomisc import bind_socket from analyzer.api.app import create_app def main () : # Аллоцируем сокет sock = bind_socket(address= '0.0.0.0' , port= 8085 , proto_name= 'http' ) user = pwd.getpwnam( 'nobody' ) os.setgid(user.pw_gid) os.setuid(user.pw_uid) app = create_app(...) run_app(app, sock=sock) if __name__ == '__main__' : main() • В конце концов вынесли создание приложения в отдельную параметризуемую функцию create_app, чтобы можно было легко создавать идентичные приложения для тестирования. Сериализация данных Все успешные ответы обработчиков будем возвращать в формате JSON. Информацию об ошибках клиентам тоже было бы удобно получать в сериализованном виде (например, чтобы увидеть, какие поля не прошли валидацию). Документация aiohttp предлагает метод json_response , который принимает объект, сериализует его в JSON и возвращает новый объект aiohttp.web.Response с заголовком Content-Type: application/json и сериализованными данными внутри. Как сериализовать данные с помощью json_response from aiohttp.web import Application, View, run_app from aiohttp.web_response import json_response class SomeView (View) : async def get (self) : return json_response({ 'hello' : 'world' }) app = Application() app.router.add_route( '*' , '/hello' , SomeView) run_app(app) Но существует и другой способ: aiohttp позволяет зарегистрировать произвольный сериализатор для определенного типа данных ответа в реестре aiohttp.PAYLOAD_REGISTRY . Например, можно указать сериализатор aiohttp.JsonPayload для объектов типа Mapping В этом случае обработчику будет достаточно вернуть объект Response с данными ответа в параметре body . aiohttp найдет сериализатор, соответствующий типу данных и сериализует ответ. Помимо того, что сериализация объектов описана в одном месте, этот подход еще и более гибкий — он позволяет реализовывать очень интересные решения (мы рассмотрим один из вариантов использования в обработчике GET /imports/$import_id/citizens ). |