You can implement the REST API service in different ways using different tools. The described solution is not the only right one, I chose the implementation and tools based on my personal experience and preferences.

• What do we do?
• What tools to choose?
• Development
  
  • Why do you need to start with setup.py?
  • How to specify dependency versions?
  • Database
    
    • We design the scheme
    • Describe the schema in SQLAlchemy
    • Customize Alembic
    • We generate migrations
  • application
    • Data serialization
    • Handlers
      • POST / imports
      • GET / imports / $ import_id / citizens
      • PATCH / imports / $ import_id / citizens / $ citizen_id
      • GET / imports / $ import_id / citizens / birthdays
      • GET / imports / $ import_id / towns / stat / percentile / age
• Testing
  • Handlers
    • GET / imports / $ import_id / citizens
    • POST / imports
    • PATCH / imports / $ import_id / citizens / $ citizen_id
    • GET / imports / $ import_id / citizens / birthdays
    • GET / imports / $ import_id / towns / stat / percentile / age
  • Migrations
• Assembly
• Ci
• Deploy
• Stress Testing
• What else can be done?
• Finally

What do we do?

• POST /imports
  Adds a new upload with data;
  
• GET /imports/$import_id/citizens
  Returns the residents of the specified discharge;
  
• PATCH /imports/$import_id/citizens/$citizen_id
  Changes information about the resident (and his relatives) in the specified unloading;
  
• GET /imports/$import_id/citizens/birthdays
  , ( ), ;
  
• GET /imports/$import_id/towns/stat/percentile/age
  50-, 75- 99- ( ) .
  

?

Development

$ python
>>> import analyzer
>>> help(analyzer)

Help on package analyzer:

NAME
    analyzer

DESCRIPTION
      REST API,    .

PACKAGE CONTENTS
    api (package)
    db (package)
    utils (package)

DATA
    __all__ = ('__author__', '__email__', '__license__', '__maintainer__',...
    __email__ = 'alvassin@yandex.ru'
    __license__ = 'MIT'
    __maintainer__ = 'Alexander Vasin'

VERSION
    0.0.1

AUTHOR
    Alexander Vasin

FILE
    /Users/alvassin/Work/backendschool2019/analyzer/__init__.py

# REST API
$ python -m analyzer.api --help

#    
$ python -m analyzer.db --help

Why do you need to start with setup.py?

setup(..., install_requires=["aiohttp", "SQLAlchemy"])

def load_requirements(fname: str) -> list:
    requirements = []
    with open(fname, 'r') as fp:
        for req in parse_requirements(fp.read()):
            extras = '[{}]'.format(','.join(req.extras)) if req.extras else ''
            requirements.append(
                '{}{}{}'.format(req.name, extras, req.specifier)
            )
    return requirements

import os
from importlib.machinery import SourceFileLoader

from pkg_resources import parse_requirements
from setuptools import find_packages, setup

module_name = 'analyzer'

# ,     (   ), 
#   __init__.py   machinery.
module = SourceFileLoader(
    module_name, os.path.join(module_name, '__init__.py')
).load_module()

def load_requirements(fname: str) -> list:
    requirements = []
    with open(fname, 'r') as fp:
        for req in parse_requirements(fp.read()):
            extras = '[{}]'.format(','.join(req.extras)) if req.extras else ''
            requirements.append(
                '{}{}{}'.format(req.name, extras, req.specifier)
            )
    return requirements

setup(
    name=module_name,
    version=module.__version__,
    author=module.__author__,
    author_email=module.__email__,
    license=module.__license__,
    description=module.__doc__,
    long_description=open('README.rst').read(),
    url='https://github.com/alvassin/backendschool2019',
    platforms='all',
    classifiers=[
        'Intended Audience :: Developers',
        'Natural Language :: Russian',
        'Operating System :: MacOS',
        'Operating System :: POSIX',
        'Programming Language :: Python',
        'Programming Language :: Python :: 3',
        'Programming Language :: Python :: 3.8',
        'Programming Language :: Python :: Implementation :: CPython'
    ],
    python_requires='>=3.8',
    packages=find_packages(exclude=['tests']),
    install_requires=load_requirements('requirements.txt'),
    extras_require={'dev': load_requirements('requirements.dev.txt')},
    entry_points={
        'console_scripts': [
            # f-strings  setup.py   - 
            # .
            #   ,     Python 3.8, 
            # source distribution       
            #   Python.     
            # .
            '{0}-api = {0}.api.__main__:main'.format(module_name),
            '{0}-db = {0}.db.__main__:main'.format(module_name)
        ]
    },
    include_package_data=True
)

#      extra- "dev"
pip install -e '.[dev]'

#      
pip install -e .

How to specify dependency versions?

• MAJOR - increases when backward incompatible changes are added;
• MINOR - Increases when adding new functionality with support for backward compatibility;
• PATCH - increases when adding bug fixes with backward compatibility support.

Database

We design the scheme

{
  "citizens": [
    {
      "citizen_id": 1,
      "town": "",
      "street": " ",
      "building": "1675",
      "apartment": 7,
      "name": "  ",
      "birth_date": "26.12.1986",
      "gender": "male",
      "relatives": [2]
    },
    {
      "citizen_id": 2,
      "town": "",
      "street": " ",
      "building": "1675",
      "apartment": 7,
      "name": "  ",
      "birth_date": "01.04.1997",
      "gender": "male",
      "relatives": [1]
    },
    {
      "citizen_id": 3,
      "town": "",
      "street": " ",
      "building": "2",
      "apartment": 11,
      "name": "  ",
      "birth_date": "23.11.1986",
      "gender": "female",
      "relatives": []
    },
    ...
  ]
}

1. The imports table consists of an automatically incrementing column import_id. It is needed to create a foreign key check in the table citizens.
   
2. The citizens table stores scalar data about the resident (all fields except information about family relationships).
   
   A pair ( import_id, citizen_id) is used as the primary key , guaranteeing the uniqueness of residents citizen_idwithin the framework import_id.
   
   A foreign key citizens.import_id -> imports.import_idensures that the field citizens.import_idcontains only existing unloads.
   
3. relations .
   
   ( ): citizens relations .
   (import_id, citizen_id, relative_id) , import_id citizen_id c relative_id.
   
   : (relations.import_id, relations.citizen_id) -> (citizens.import_id, citizens.citizen_id) (relations.import_id, relations.relative_id) -> (citizens.import_id, citizens.citizen_id), , citizen_id relative_id .

Describe the schema in SQLAlchemy

# analyzer/db/schema.py
from sqlalchemy import MetaData

convention = {
    'all_column_names': lambda constraint, table: '_'.join([
        column.name for column in constraint.columns.values()
    ]),

    #  
    'ix': 'ix__%(table_name)s__%(all_column_names)s',

    #   
    'uq': 'uq__%(table_name)s__%(all_column_names)s',

    #  CHECK-constraint-
    'ck': 'ck__%(table_name)s__%(constraint_name)s',

    #   
    'fk': 'fk__%(table_name)s__%(all_column_names)s__%(referred_table_name)s',

    #   
    'pk': 'pk__%(table_name)s'
}
metadata = MetaData(naming_convention=convention)

# analyzer/db/schema.py
from enum import Enum, unique

from sqlalchemy import (
    Column, Date, Enum as PgEnum, ForeignKey, ForeignKeyConstraint, Integer,
    String, Table
)


@unique
class Gender(Enum):
    female = 'female'
    male = 'male'


imports_table = Table(
    'imports',
    metadata,
    Column('import_id', Integer, primary_key=True)
)

citizens_table = Table(
    'citizens',
    metadata,
    Column('import_id', Integer, ForeignKey('imports.import_id'),
           primary_key=True),
    Column('citizen_id', Integer, primary_key=True),
    Column('town', String, nullable=False, index=True),
    Column('street', String, nullable=False),
    Column('building', String, nullable=False),
    Column('apartment', Integer, nullable=False),
    Column('name', String, nullable=False),
    Column('birth_date', Date, nullable=False),
    Column('gender', PgEnum(Gender, name='gender'), nullable=False),
)

relations_table = Table(
    'relations',
    metadata,
    Column('import_id', Integer, primary_key=True),
    Column('citizen_id', Integer, primary_key=True),
    Column('relative_id', Integer, primary_key=True),
    ForeignKeyConstraint(
        ('import_id', 'citizen_id'),
        ('citizens.import_id', 'citizens.citizen_id')
    ),
    ForeignKeyConstraint(
        ('import_id', 'relative_id'),
        ('citizens.import_id', 'citizens.citizen_id')
    ),
)

Customize Alembic

1. Install Package: pip install alembic
2. Initialize Alembic: cd analyzer && alembic init db/alembic.
   
   This command will create a configuration file analyzer/alembic.iniand a folder analyzer/db/alembicwith the following contents:
   
   • env.py- Called every time you start Alembic. Connects to Alembic registry sqlalchemy.MetaDatawith a description of the desired state of the database and contains instructions for starting migrations.
     
   • script.py.mako - the template on the basis of which migrations are generated.
   • versions - the folder in which Alembic will search (and generate) migrations.
3. Specify the database address in the alembic.ini file:
   
   

   ; analyzer/alembic.ini
   [alembic] 
   sqlalchemy.url = postgresql://user:hackme@localhost/analyzer

4. Specify a description of the desired state of the database (registry sqlalchemy.MetaData) so that Alembic can generate migrations automatically:
   
   

   # analyzer/db/alembic/env.py
   from analyzer.db import schema
   target_metadata = schema.metadata

1. The utility alembicsearches alembic.iniin the current working directory. You alembic.inican specify the path to the command line argument, but this is inconvenient: I want to be able to call the command from any folder without additional parameters.
2. To configure Alembic to work with a specific database, you need to change the file alembic.ini. It would be much more convenient to specify the database settings for the environment variable and / or a command line argument, for example --pg-url.
3. The name of the utility alembicdoes not correlate very well with the name of our service (and the user may actually not have Python at all and know nothing about Alembic). It would be much more convenient for the end user if all executable commands of the service had a common prefix, for example analyzer-*.

• Alembic uses a standard module to process command line arguments argparse. It allows you to add an optional argument --pg-urlwith a default value from an environment variable ANALYZER_PG_URL.
  
  
  The code

  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()

• The path to the file alembic.inican be calculated relative to the location of the executable file, and not the current working directory of the user.
  
  
  The code

  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()

from setuptools import setup

setup(..., entry_points={
    'console_scripts': [
        'analyzer-db = analyzer.db.__main__:main'
    ]
})

$ pip install -e '.[dev]'

We generate migrations

$ 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/d5f704ed4610_initial.py ...  done

• The user data types specified in the created tables are created automatically (in our case - gender), but the code to delete them is downgradenot generated. If you apply, roll back, and then apply the migration again, this will cause an error because the specified data type already exists.
  
  
  Delete the gender data type in the downgrade method

  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())

• In the method, downgradesome actions can sometimes be removed (if we delete the whole table, you can not delete its indexes separately):
  
  
  for instance

  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

application

import logging

from aiohttp import web


def main():
    #  
    logging.basicConfig(level=logging.DEBUG)

    #  
    app = web.Application()

    #  
    app.router.add_route(...)

    #  
    web.run_app(app)

• How to configure the application? At a minimum, you must specify the host and port for connecting clients, as well as information for connecting to the database.
  
  I really like to solve this problem with the help of the module ConfigArgParse: it extends the standard one argparseand allows using command line arguments, environment variables (indispensable for configuring Docker containers) and even configuration files (as well as combining these methods) for configuration. Using ConfigArgParseit, you can also validate the values ​​of application configuration parameters.
  
  
  An example of processing parameters using 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/ .
  
  9 , 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, .

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)

from types import MappingProxyType
from typing import Mapping

from aiohttp import PAYLOAD_REGISTRY, JsonPayload
from aiohttp.web import run_app, Application, Response, View

PAYLOAD_REGISTRY.register(JsonPayload, (Mapping, MappingProxyType))


class SomeView(View):
    async def get(self):
        return Response(body={'hello': 'world'})


app = Application()
app.router.add_route('*', '/hello', SomeView)
run_app(app)

import json
from datetime import date
from functools import partial, singledispatch
from typing import Any

from aiohttp.payload import JsonPayload as BaseJsonPayload
from aiohttp.typedefs import JSONEncoder

@singledispatch
def convert(value):
    raise NotImplementedError(f'Unserializable value: {value!r}')


@convert.register(Record)
def convert_asyncpg_record(value: Record):
    """
        , 
    asyncpg
    """
    return dict(value)


@convert.register(date)
def convert_date(value: date):
    """
       date      —  
      .     
      ..
    """
    return value.strftime('%d.%m.%Y')
    
 
dumps = partial(json.dumps, default=convert)


class JsonPayload(BaseJsonPayload):
    def __init__(self,
                 value: Any,
                 encoding: str = 'utf-8',
                 content_type: str = 'application/json',
                 dumps: JSONEncoder = dumps,
                 *args: Any,
                 **kwargs: Any) -> None:
        super().__init__(value, encoding, content_type, dumps, *args, **kwargs)

Handlers

from aiohttp.web_urldispatcher import View
from asyncpgsa import PG


class BaseView(View):
    URL_PATH: str

    @property
    def pg(self) -> PG:
        return self.request.app['pg']

POST / imports

• Data type, format and availability of all required fields;
• Lack of unfamiliar fields;
• Date of birth must be indicated in the format DD.MM.YYYYand cannot be of any significance from the future
• The list of relatives of each resident must contain unique identifiers of residents existing in this upload.

• citizen_id each resident within the unloading should be unique;
• Family ties should be two-way (if resident # 1 has a resident # 2 in the list of relatives, then resident # 2 must also have a relative # 1).

GET / imports / $ import_id / citizens

from aiohttp.web_exceptions import HTTPNotFound
from sqlalchemy import select, exists

from analyzer.db.schema import imports_table


class BaseImportView(BaseView):
    @property
    def import_id(self):
        return int(self.request.match_info.get('import_id'))

    async def check_import_exists(self):
        query = select([
            exists().where(imports_table.c.import_id == self.import_id)
        ])
        if not await self.pg.fetchval(query):
            raise HTTPNotFound()

1. A SelectQuerytype object AsyncIterablethat returns records from the database. At the first call, it connects to the database, opens a transaction and creates a cursor; during further iteration, it returns records from the database. It is returned by the handler.
   
   
   SelectQuery Code

   from collections import AsyncIterable
   from asyncpgsa.transactionmanager import ConnectionTransactionContextManager
   from sqlalchemy.sql import Select
   
   
   class SelectQuery(AsyncIterable):
       """
       ,     PostgreSQL   
       ,  ,    
       """
       PREFETCH = 500
   
       __slots__ = (
           'query', 'transaction_ctx', 'prefetch', 'timeout'
       )
   
       def __init__(self, query: Select,
                    transaction_ctx: ConnectionTransactionContextManager,
                    prefetch: int = None,
                    timeout: float = None):
           self.query = query
           self.transaction_ctx = transaction_ctx
           self.prefetch = prefetch or self.PREFETCH
           self.timeout = timeout
   
       async def __aiter__(self):
           async with self.transaction_ctx as conn:
               cursor = conn.cursor(self.query, prefetch=self.prefetch,
                                    timeout=self.timeout)
               async for row in cursor:
                   yield row
   

2. A serializer AsyncGenJSONListPayloadthat can iterate over asynchronous generators, serialize data from an asynchronous generator to JSON and send data to clients in parts. It is registered aiohttp.PAYLOAD_REGISTRYas a serializer of objects AsyncIterable.
   
   
   AsyncGenJSONListPayload Code

   import json
   from functools import partial
   
   from aiohttp import Payload
   
   
   # ,    JSON  asyncpg.Record  datetime.date
   dumps = partial(json.dumps, default=convert, ensure_ascii=False)
   
   
   class AsyncGenJSONListPayload(Payload):
       """
          AsyncIterable,     
        JSON   
       """
       def __init__(self, value, encoding: str = 'utf-8',
                    content_type: str = 'application/json',
                    root_object: str = 'data',
                    *args, **kwargs):
           self.root_object = root_object
           super().__init__(value, content_type=content_type, encoding=encoding,
                            *args, **kwargs)
   
       async def write(self, writer):
           #  
           await writer.write(
               ('{"%s":[' % self.root_object).encode(self._encoding)
           )
   
           first = True
           async for row in self._value:
               #      
               if not first:
                   await writer.write(b',')
               else:
                   first = False
   
               await writer.write(dumps(row).encode(self._encoding))
   
           #  
           await writer.write(b']}')

# analyzer/api/handlers/citizens.py
from aiohttp.web_response import Response
from aiohttp_apispec import docs, response_schema

from analyzer.api.schema import CitizensResponseSchema
from analyzer.db.schema import citizens_table as citizens_t
from analyzer.utils.pg import SelectQuery

from .query import CITIZENS_QUERY
from .base import BaseImportView


class CitizensView(BaseImportView):
    URL_PATH = r'/imports/{import_id:\d+}/citizens'

    @docs(summary='    ')
    @response_schema(CitizensResponseSchema())
    async def get(self):
        await self.check_import_exists()

        query = CITIZENS_QUERY.where(
            citizens_t.c.import_id == self.import_id
        )
        body = SelectQuery(query, self.pg.transaction())
        return Response(body=body)

PATCH / imports / $ import_id / citizens / $ citizen_id

• The type and format of the data for the specified fields.
• Date of birth. It must be specified in a format DD.MM.YYYYand cannot be of significance from the future.
• A list of relatives of each resident. It must have unique identifiers for residents.

GET / imports / $ import_id / citizens / birthdays

1. Get data for residents with relatives from the database, and on the Python side, aggregate data by month and generate lists for those months for which there is no data in the database.
2. Compile a json request in the database and add stubs for the missing months.

GET / imports / $ import_id / towns / stat / percentile / age

1. Get the age of residents from the database, grouped by city, and then on the Python side calculate the percentiles using numpy (which is specified as a reference in the task) and round up to two decimal places.
2. PostgreSQL: percentile_cont , SQL-, numpy .

Testing

import os
import uuid

import pytest
from sqlalchemy import create_engine
from sqlalchemy_utils import create_database, drop_database
from yarl import URL

from analyzer.utils.pg import DEFAULT_PG_URL

PG_URL = os.getenv('CI_ANALYZER_PG_URL', DEFAULT_PG_URL)


@pytest.fixture
def postgres():
    tmp_name = '.'.join([uuid.uuid4().hex, 'pytest'])
    tmp_url = str(URL(PG_URL).with_path(tmp_name))
    create_database(tmp_url)

    try:
        #      postgres  -
        yield tmp_url
    finally:
        drop_database(tmp_url)


def test_db(postgres):
    """
     ,  PostgreSQL
    """
    engine = create_engine(postgres)
    assert engine.execute('SELECT 1').scalar() == 1
    engine.dispose()

from types import SimpleNamespace

import pytest

from analyzer.utils.pg import make_alembic_config


@pytest.fixture()
def alembic_config(postgres):
    cmd_options = SimpleNamespace(config='alembic.ini', name='alembic',
                                  pg_url=postgres, raiseerr=False, x=None)
    return make_alembic_config(cmd_options)

Handlers

from alembic.command import upgrade

@pytest.fixture
async def migrated_postgres(alembic_config, postgres):
    upgrade(alembic_config, 'head')
    #  DSN  ,    
    return postgres

@pytest.fixture
def arguments(aiomisc_unused_port, migrated_postgres):
    return parser.parse_args(
        [
            '--log-level=debug',
            '--api-address=127.0.0.1',
            f'--api-port={aiomisc_unused_port}',
            f'--pg-url={migrated_postgres}'
        ]
    )

from analyzer.api.app import create_app

@pytest.fixture
async def api_client(aiohttp_client, arguments):
    app = create_app(arguments)
    client = await aiohttp_client(app, server_kwargs={
        'port': arguments.api_port
    })

    try:
        yield client
    finally:
        await client.close()

1. postgres ( migrated_postgres).
2. alembic_config Alembic, ( migrated_postgres).
3. migrated_postgres ( arguments).
4. aiomisc_unused_port ( arguments).
5. arguments ( api_client).
6. api_client .
7. .
8. api_client .
9. postgres .

GET / imports / $ import_id / citizens

• Unloading with several relatives. Checks that for each resident a list with identifiers of relatives will be correctly formed.
• Unloading with one resident without relatives. Checks that the field relativesis an empty list (due LEFT JOINto the SQL query, the list of relatives may be equal [None]).
• Unloading with a resident who is a relative of himself.
• Empty unloading. Checks that the handler allows adding empty unloading and does not crash with an error.

import pytest

from analyzer.utils.testing import generate_citizen

datasets = [
    #    
    [
        generate_citizen(citizen_id=1, relatives=[2, 3]),
        generate_citizen(citizen_id=2, relatives=[1]),
        generate_citizen(citizen_id=3, relatives=[1])
    ],

    #   
    [
        generate_citizen(relatives=[])
    ],

    #   ,    
    [
        generate_citizen(citizen_id=1, name='', gender='male',
                         birth_date='17.02.2020', relatives=[1])
    ],

    #  
    [],
]


@pytest.mark.parametrize('dataset', datasets)
async def test_get_citizens(api_client, dataset):
    """
        4 ,    
    """

from typing import Iterable, Mapping

def normalize_citizen(citizen):
    """
         
    """
    return {**citizen, 'relatives': sorted(citizen['relatives'])}


def compare_citizens(left: Mapping, right: Mapping) -> bool:
    """
      
    """
    return normalize_citizen(left) == normalize_citizen(right)


def compare_citizen_groups(left: Iterable, right: Iterable) -> bool:
    """
          ,   
      
    """
    left = [normalize_citizen(citizen) for citizen in left]
    left.sort(key=lambda citizen: citizen['citizen_id'])

    right = [normalize_citizen(citizen) for citizen in right]
    right.sort(key=lambda citizen: citizen['citizen_id'])
    return left == right

POST / imports

• Correct data, expected to be successfully added to the database.
  
  
  • ( ).
    
    . , , insert , .
  • ( , ).
    
    , .
    
  • .
    
    , . :)
    
  • 
    
    , aiohttp PostgreSQL 32 767 ( ).
    
  • Empty unloading
    
    The handler should take into account such a case and not fall, trying to perform empty insert into the table with the inhabitants.
    
  
  
• Data with errors, expect an HTTP response of 400: Bad Request.
  
  
  • Date of birth is incorrect (future tense).
  • citizen_id is not unique within the upload.
  • A kinship is indicated incorrectly (there is only from one resident to another, but there is no feedback).
  • The resident has a non-existent relative in the unloading.
  • Family ties are not unique.

PATCH / imports / $ import_id / citizens / $ citizen_id

• The field is specified, but has an incorrect data type and / or format
• The date of birth is incorrect (future time).
• The field relativescontains a relative that does not exist in the unloading.

GET / imports / $ import_id / citizens / birthdays

• An unloading in which a resident has one relative in one month and two relatives in another.
• Unloading with one resident without relatives. Checks that the handler does not take it into account in the calculations.
• Empty unloading. Checks that the handler will not fail and will return the correct dictionary with 12 months in the response.
• Unloading with a resident who is a relative of himself. Checks that a resident will buy a gift for the month of his birth.

GET / imports / $ import_id / towns / stat / percentile / age

from unittest.mock import patch

import pytz

CURRENT_DATE = datetime(2020, 2, 17, tzinfo=pytz.utc)


@patch('analyzer.api.handlers.TownAgeStatView.CURRENT_DATE', new=CURRENT_DATE)
async def test_get_ages(...):
    ...

• , ( — 364 ). , .
• , ( — ). — , , 1 .
• . .

# Python 3
round(2.5)
> 2

-- PostgreSQL
SELECT ROUND(2.5)
> 3

import pytz

from analyzer.utils.testing import generate_citizen


CURRENT_DATE = datetime(2020, 2, 17, tzinfo=pytz.utc)

def age2date(years: int, days: int = 0, base_date=CURRENT_DATE) -> str:
    birth_date = copy(base_date).replace(year=base_date.year - years)
    birth_date -= timedelta(days=days)
    return birth_date.strftime(BIRTH_DATE_FORMAT)

#    ?  ,     ?
generate_citizen(birth_date='17.02.2009')

#   11       
generate_citizen(birth_date=age2date(years=11))

An interesting fact: when I wrote this test on February 29, 2020, I suddenly stopped generating unloads with residents due to a bug in Faker (2020 is a leap year, and other years that Faker chose were not always leap years in them too was not February 29th). Remember to record dates and test edge cases!

Migrations

• downgrade ( , , ).
  
  , (--): , — .
  
• C .
• ( ).

Assembly

1. Using a “heavy” image snakepacker/python:all(~ 1 GB, ~ 500 MB compressed), create a virtual environment, install all the dependencies and the application package in it. This image is needed exclusively for assembly, it can contain a compiler, all the necessary libraries and files with headers.
   
   

   FROM snakepacker/python:all as builder
   
   #   
   RUN python3.8 -m venv /usr/share/python3/app
   
   #  source distribution     
   COPY dist/ /mnt/dist/
   RUN /usr/share/python3/app/bin/pip install /mnt/dist/*

2. We copy the finished virtual environment into a “light” image snakepacker/python:3.8(~ 100 MB, compressed ~ 50 MB), which contains only the interpreter of the required version of Python.
   
   Important: in a virtual environment, absolute paths are used, so it must be copied to the same address at which it was assembled in the collector container.
   
   

   FROM snakepacker/python:3.8 as api
   
   #       builder
   COPY --from=builder /usr/share/python3/app /usr/share/python3/app
   
   #  ,     
   # 
   RUN ln -snf /usr/share/python3/app/bin/analyzer-* /usr/local/bin/
   
   #        
   CMD ["analyzer-api"]

###############      ################
#  — «» (~1 ,    ~500 )    
#    
FROM snakepacker/python:all as builder

#      pip
RUN python3.8 -m venv /usr/share/python3/app
RUN /usr/share/python3/app/bin/pip install -U pip

#   ,  .   
# Docker   ,  requirements.txt  
COPY requirements.txt /mnt/
RUN /usr/share/python3/app/bin/pip install -Ur /mnt/requirements.txt

#  source distribution     
COPY dist/ /mnt/dist/
RUN /usr/share/python3/app/bin/pip install /mnt/dist/* \
    && /usr/share/python3/app/bin/pip check

###########################   ############################
#    «» (~100 ,    ~50 )   Python
FROM snakepacker/python:3.8 as api

#         builder
COPY --from=builder /usr/share/python3/app /usr/share/python3/app

#  ,     
# 
RUN ln -snf /usr/share/python3/app/bin/analyzer-* /usr/local/bin/

#        
CMD ["analyzer-api"]

Ci

name: CI

# Workflow      
#   -  master
on:
  push:
    branches: [ master ]
  pull_request:
    branches: [ master ]

jobs:
  #       workflow
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: docker://postgres
        ports:
          - 5432:5432
        env:
          POSTGRES_USER: user
          POSTGRES_PASSWORD: hackme
          POSTGRES_DB: analyzer

    steps:
      - uses: actions/checkout@v2
      - name: test
        uses: docker://snakepacker/python:all
        env:
          CI_ANALYZER_PG_URL: postgresql://user:hackme@postgres/analyzer
        with:
          args: /bin/bash -c "pip install -U '.[dev]' && pylama && wait-for-port postgres:5432 && pytest -vv --cov=analyzer --cov-report=term-missing tests"

  #    Docker-  
  publish:
    #        master
    if: github.event_name == 'push' && github.ref == 'refs/heads/master'
    # ,   test   
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: sdist
        uses: docker://snakepacker/python:all
        with:
          args: make sdist

      - name: build-push
        uses: docker/build-push-action@v1
        with:
          username: ${{ secrets.REGISTRY_LOGIN }}
          password: ${{ secrets.REGISTRY_TOKEN }}
          repository: alvassin/backendschool2019
          target: api
          tags: 0.0.1, latest

Deploy

[Unit]
Description=%i service with docker compose
Requires=docker.service
After=docker.service

[Service]
Type=oneshot
RemainAfterExit=true
WorkingDirectory=/etc/docker/compose/%i
ExecStart=/usr/local/bin/docker-compose up -d --remove-orphans
ExecStop=/usr/local/bin/docker-compose down

[Install]
WantedBy=multi-user.target

---

- name: Gathering facts
  hosts: all
  become: yes
  gather_facts: yes

- name: Install docker
  hosts: docker
  become: yes
  gather_facts: no
  roles:
    - docker

- name: Install analyzer
  hosts: api
  become: yes
  gather_facts: no
  roles:
    - analyzer

[api]
130.193.51.154

[docker:children]
api

[api:vars]
analyzer_image = alvassin/backendschool2019
analyzer_pg_user = user
analyzer_pg_password = hackme
analyzer_pg_dbname = analyzer

$ ansible-playbook -i hosts.ini deploy.yml

About stress testing

, , . , - . : , — , 10 . , (, , CI-): .

, , , 10 . ? , , . , , .

RPS, : . , , import_id, POST /imports . .

, Python 3, Locust.

, locustfile.py locust. - .

Locust . , .
self.round .

locustfile.py

# locustfile.py
import logging
from http import HTTPStatus

from locust import HttpLocust, constant, task, TaskSet
from locust.exception import RescheduleTask

from analyzer.api.handlers import (
    CitizenBirthdaysView, CitizensView, CitizenView, TownAgeStatView
)
from analyzer.utils.testing import generate_citizen, generate_citizens, url_for


class AnalyzerTaskSet(TaskSet):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.round = 0

    def make_dataset(self):
        citizens = [
            #     .   
            # PATCH-  relatives    
            # ,     - 
            # (     ,    
            # ).
            generate_citizen(citizen_id=1, relatives=[2]),
            generate_citizen(citizen_id=2, relatives=[1]),
            *generate_citizens(citizens_num=9998, relations_num=1000,
                               start_citizen_id=3)
        ]
        return {citizen['citizen_id']: citizen for citizen in citizens}

    def request(self, method, path, expected_status, **kwargs):
        with self.client.request(
                method, path, catch_response=True, **kwargs
        ) as resp:
            if resp.status_code != expected_status:
                resp.failure(f'expected status {expected_status}, '
                             f'got {resp.status_code}')
            logging.info(
                'round %r: %s %s, http status %d (expected %d), took %rs',
                self.round, method, path, resp.status_code, expected_status,
                resp.elapsed.total_seconds()
            )
            return resp

    def create_import(self, dataset):
        resp = self.request('POST', '/imports', HTTPStatus.CREATED,
                            json={'citizens': list(dataset.values())})
        if resp.status_code != HTTPStatus.CREATED:
            raise RescheduleTask
        return resp.json()['data']['import_id']

    def get_citizens(self, import_id):
        url = url_for(CitizensView.URL_PATH, import_id=import_id)
        self.request('GET', url, HTTPStatus.OK,
                     name='/imports/{import_id}/citizens')

    def update_citizen(self, import_id):
        url = url_for(CitizenView.URL_PATH, import_id=import_id, citizen_id=1)
        self.request('PATCH', url, HTTPStatus.OK,
                     name='/imports/{import_id}/citizens/{citizen_id}',
                     json={'relatives': [i for i in range(3, 10)]})

    def get_birthdays(self, import_id):
        url = url_for(CitizenBirthdaysView.URL_PATH, import_id=import_id)
        self.request('GET', url, HTTPStatus.OK,
                     name='/imports/{import_id}/citizens/birthdays')

    def get_town_stats(self, import_id):
        url = url_for(TownAgeStatView.URL_PATH, import_id=import_id)
        self.request('GET', url, HTTPStatus.OK,
                     name='/imports/{import_id}/towns/stat/percentile/age')

    @task
    def workflow(self):
        self.round += 1
        dataset = self.make_dataset()

        import_id = self.create_import(dataset)
        self.get_citizens(import_id)
        self.update_citizen(import_id)
        self.get_birthdays(import_id)
        self.get_town_stats(import_id)


class WebsiteUser(HttpLocust):
    task_set = AnalyzerTaskSet
    wait_time = constant(1)

100 c , , :



, ( — 95 , — ). .



— Ansible ~20.15 ~20.30 Locust.

What else can be done?

* * *

Acknowledgments