Vous pouvez implémenter le service API REST de différentes manières à l'aide de différents outils. La solution décrite n'est pas la seule bonne, j'ai choisi la mise en œuvre et les outils en fonction de mon expérience et de mes préférences personnelles.

• Qu'est-ce qu'on fait?
• Quels outils choisir?
• Développement
  
  • Pourquoi avez-vous besoin de commencer avec setup.py?
  • Comment spécifier des versions de dépendances?
  • Base de données
    
    • Nous concevons le schéma
    • Décrire le schéma dans SQLAlchemy
    • Personnaliser Alembic
    • Nous générons des migrations
  • application
    • Sérialisation des données
    • Gestionnaires
      • POST / importations
      • GET / importations / $ import_id / citoyens
      • PATCH / imports / $ import_id / citoyens / $ citizen_id
      • GET / importations / $ import_id / citoyens / anniversaires
      • GET / imports / $ import_id / towns / stat / percentile / age
• Essai
  • Gestionnaires
    • GET / importations / $ import_id / citoyens
    • POST / importations
    • PATCH / imports / $ import_id / citoyens / $ citizen_id
    • GET / importations / $ import_id / citoyens / anniversaires
    • GET / imports / $ import_id / towns / stat / percentile / age
  • Migrations
• Assemblée
• Ci
• Déployer
• Test de résistance
• Que peut-on faire d'autre?
• finalement

Qu'est-ce qu'on fait?

• POST /imports
  Ajoute un nouveau téléchargement avec des données;
  
• GET /imports/$import_id/citizens
  Renvoie les résidents de la sortie spécifiée;
  
• PATCH /imports/$import_id/citizens/$citizen_id
  Modifie les informations sur le résident (et ses proches) dans le déchargement spécifié;
  
• GET /imports/$import_id/citizens/birthdays
  , ( ), ;
  
• GET /imports/$import_id/towns/stat/percentile/age
  50-, 75- 99- ( ) .
  

?

Développement

$ 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

Pourquoi avez-vous besoin de commencer avec 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 .

Comment spécifier des versions de dépendances?

• MAJOR - augmente lorsque des modifications incompatibles vers l'arrière sont ajoutées;
• MINOR - Augmente lors de l'ajout de nouvelles fonctionnalités avec prise en charge de la compatibilité descendante;
• PATCH - augmente lors de l'ajout de corrections de bogues avec prise en charge de la compatibilité descendante.

Base de données

Nous concevons le schéma

{
  "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. Le tableau des importations se compose d'une colonne à incrémentation automatique import_id. Il est nécessaire de créer une vérification de clé étrangère dans le tableau citizens.
   
2. La table des citoyens stocke des données scalaires sur le résident (tous les champs sauf les informations sur les relations familiales).
   
   Une paire ( import_id, citizen_id) est utilisée comme clé primaire , garantissant l'unicité des résidents citizen_iddans le cadre import_id.
   
   Une clé étrangère citizens.import_id -> imports.import_idgarantit que le champ citizens.import_idcontient uniquement les déchargements existants.
   
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 .

Décrire le schéma dans 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')
    ),
)

Personnaliser Alembic

1. Installer le paquet: pip install alembic
2. Initialize Alambic: cd analyzer && alembic init db/alembic.
   
   Cette commande créera un fichier de configuration analyzer/alembic.iniet un dossier analyzer/db/alembicavec le contenu suivant:
   
   • env.py- Appelé chaque fois que vous démarrez Alembic. Se connecte au registre Alembic sqlalchemy.MetaDataavec une description de l'état souhaité de la base de données et contient des instructions pour démarrer les migrations.
     
   • script.py.mako - le modèle sur la base duquel les migrations sont générées.
   • versions - le dossier dans lequel Alembic recherchera (et générera) les migrations.
3. Spécifiez l'adresse de la base de données dans le fichier alembic.ini:
   
   

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

4. Spécifiez une description de l'état souhaité de la base de données (registre sqlalchemy.MetaData) afin qu'Alembic puisse générer automatiquement des migrations:
   
   

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

1. L'utilitaire alembicrecherche alembic.inidans le répertoire de travail actuel. Vous alembic.inipouvez spécifier le chemin d'accès à l' argument de ligne de commande, mais cela n'est pas pratique: je veux pouvoir appeler la commande à partir de n'importe quel dossier sans paramètres supplémentaires.
2. Pour configurer Alembic pour fonctionner avec une base de données spécifique, vous devez modifier le fichier alembic.ini. Il serait beaucoup plus pratique de spécifier les paramètres de base de données pour la variable d'environnement et / ou un argument de ligne de commande, par exemple --pg-url.
3. Le nom de l'utilitaire alembicne correspond pas très bien au nom de notre service (et l'utilisateur peut en fait ne pas avoir du tout Python et ne rien savoir sur Alembic). Il serait beaucoup plus pratique pour l'utilisateur final que toutes les commandes exécutables du service aient un préfixe commun, par exemple analyzer-*.

• Alembic utilise un module standard pour traiter les arguments de ligne de commande argparse. Il vous permet d'ajouter un argument facultatif --pg-urlavec une valeur par défaut à partir d'une variable d'environnement ANALYZER_PG_URL.
  
  
  Le 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()

• Le chemin d'accès au fichier alembic.inipeut être calculé par rapport à l'emplacement du fichier exécutable et non au répertoire de travail actuel de l'utilisateur.
  
  
  Le 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]'

Nous générons des 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

• Les types de données utilisateur spécifiés dans les tables créées sont créés automatiquement (dans notre cas - gender), mais le code pour les supprimer n'est downgradepas généré. Si vous appliquez, annulez, puis appliquez à nouveau la migration, cela provoquera une erreur car le type de données spécifié existe déjà.
  
  
  Supprimer le type de données de genre dans la méthode de rétrogradation

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

• Dans la méthode, downgradecertaines actions peuvent parfois être supprimées (si nous supprimons la table entière, vous ne pouvez pas supprimer ses index séparément):
  
  
  par exemple

  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)

• Comment configurer l'application? Au minimum, vous devez spécifier l'hôte et le port de connexion des clients, ainsi que les informations de connexion à la base de données.
  
  J'aime vraiment résoudre ce problème avec l'aide du module ConfigArgParse: il étend le standard argparseet permet d'utiliser des arguments de ligne de commande, des variables d'environnement (indispensables pour configurer les conteneurs Docker) et même des fichiers de configuration (ainsi que la combinaison de ces méthodes) pour la configuration. En l'utilisant ConfigArgParse, vous pouvez également valider les valeurs des paramètres de configuration d'application.
  
  
  Un exemple de traitement des paramètres à l'aide de 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)

Gestionnaires

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 / importations

• Type de données, format et disponibilité de tous les champs obligatoires;
• Manque de champs inconnus;
• La date de naissance doit être indiquée dans le format DD.MM.YYYYet ne peut avoir aucune signification pour l'avenir;
• La liste des proches de chaque résident doit contenir des identifiants uniques des résidents présents dans ce téléchargement.

• citizen_id chaque résident du déchargement doit être unique;
• Les liens familiaux doivent être à double sens (si le résident n ° 1 a un résident n ° 2 dans la liste des parents, le résident n ° 2 doit également avoir un parent n ° 1).

GET / importations / $ import_id / citoyens

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. Un objet SelectQuerytype AsyncIterablequi renvoie des enregistrements de la base de données. Au premier appel, il se connecte à la base de données, ouvre une transaction et crée un curseur; lors d'une nouvelle itération, il renvoie les enregistrements de la base de données. Il est retourné par le gestionnaire.
   
   
   Code SelectQuery

   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. Un sérialiseur AsyncGenJSONListPayloadqui peut parcourir les générateurs asynchrones, sérialiser les données d'un générateur asynchrone vers JSON et envoyer des données aux clients en plusieurs parties. Il est enregistré aiohttp.PAYLOAD_REGISTRYcomme sérialiseur d'objets AsyncIterable.
   
   
   Code AsyncGenJSONListPayload

   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 / citoyens / $ citizen_id

• Le type et le format des données pour les champs spécifiés.
• Date de naissance. Il doit être spécifié dans un format DD.MM.YYYYet ne peut être significatif à l'avenir.
• Une liste des proches de chaque résident. Il doit avoir des identifiants uniques pour les résidents.

GET / importations / $ import_id / citoyens / anniversaires

1. Obtenez des données pour les résidents avec des proches dans la base de données et du côté Python, agrégez les données par mois et générez des listes pour les mois pour lesquels il n'y a pas de données dans la base de données.
2. Compilez une requête json dans la base de données et ajoutez des stubs pour les mois manquants.

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

1. Obtenez l'âge des résidents à partir de la base de données, regroupés par ville, puis du côté Python, calculez les centiles en utilisant numpy (qui est spécifié comme référence dans la tâche) et arrondissez à deux décimales.
2. PostgreSQL: percentile_cont , SQL-, numpy .

Essai

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)

Gestionnaires

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 / importations / $ import_id / citoyens

• Déchargement avec plusieurs proches. Vérifie que pour chaque résident une liste avec les identifiants des proches sera correctement constituée.
• Déchargement avec un résident sans parents. Vérifie que le champ relativesest une liste vide (en raison LEFT JOINde la requête SQL, la liste des parents peut être égale [None]).
• Déchargement avec un résident qui est un parent de lui-même.
• Déchargement vide. Vérifie que le gestionnaire autorise l'ajout de déchargement vide et ne se bloque pas avec une erreur.

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 / importations

• Données correctes, censées être ajoutées avec succès à la base de données.
  
  
  • ( ).
    
    . , , insert , .
  • ( , ).
    
    , .
    
  • .
    
    , . :)
    
  • 
    
    , aiohttp PostgreSQL 32 767 ( ).
    
  • Déchargement vide
    
    Le gestionnaire doit prendre en compte un tel cas et ne pas tomber, en essayant d'effectuer une insertion vide dans la table avec les habitants.
    
  
  
• Données avec des erreurs, attendez-vous à une réponse HTTP de 400: Bad Request.
  
  
  • La date de naissance est incorrecte (futur).
  • citizen_id n'est pas unique dans le téléchargement.
  • Une parenté est indiquée de manière incorrecte (il n'y a qu'un résident à l'autre, mais il n'y a pas de retour).
  • Le résident a un parent inexistant au déchargement.
  • Les liens familiaux ne sont pas uniques.

PATCH / imports / $ import_id / citoyens / $ citizen_id

• Le champ est spécifié, mais a un type et / ou un format de données incorrects
• La date de naissance est incorrecte (heure future).
• Le champ relativescontient un parent qui n'existe pas dans le déchargement.

GET / importations / $ import_id / citoyens / anniversaires

• Un déchargement dans lequel un résident a un parent en un mois et deux parents en un autre.
• Déchargement avec un résident sans parents. Vérifie que le gestionnaire n'en tient pas compte dans les calculs.
• Déchargement vide. Vérifie que le gestionnaire n'échouera pas et renverra le dictionnaire correct avec 12 mois dans la réponse.
• Déchargement avec un résident qui est un parent de lui-même. Vérifie qu'un résident achètera un cadeau pour le mois de sa naissance.

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

• Déchargement avec plusieurs résidents dont l'anniversaire est demain (âge - plusieurs années et 364 jours). Vérifie que le processeur utilise uniquement le nombre d'années complètes dans les calculs.
• Déchargement avec un résident dont l'anniversaire est aujourd'hui (âge - exactement quelques années). Il vérifie le cas régional - l'âge d'un résident dont l'anniversaire est aujourd'hui ne doit pas être calculé comme réduit d'un an.
• Déchargement vide. Le gestionnaire ne doit pas tomber dessus.

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

Un fait intéressant: lorsque j'ai écrit ce test le 29 février 2020, j'ai soudainement cessé de générer des décharges avec les résidents en raison d'un bug dans Faker (2020 est une année bissextile, et d'autres années que Faker a choisies n'étaient pas toujours des années bissextiles aussi) n'était pas le 29 février). N'oubliez pas d'enregistrer les dates et de tester les cas limites!

Migrations

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

Assemblée

1. À l'aide d'une image «lourde» snakepacker/python:all(~ 1 Go, ~ 500 Mo compressés), créez un environnement virtuel, installez-y toutes les dépendances et le package d'application. Cette image est nécessaire exclusivement pour l'assemblage, elle peut contenir un compilateur, toutes les bibliothèques et fichiers nécessaires avec des en-têtes.
   
   

   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. Nous copions l'environnement virtuel fini dans une image «légère» snakepacker/python:3.8(~ 100 Mo, compressée ~ 50 Mo), qui ne contient que l'interpréteur de la version requise de Python.
   
   Important: dans un environnement virtuel, des chemins absolus sont utilisés, il doit donc être copié à la même adresse à laquelle il a été assemblé dans le conteneur collecteur.
   
   

   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

Déployer

[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

À propos des tests de résistance

, , . , - . : , — , 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.

Que peut-on faire d'autre?

* * *

Remerciements