Sie können den REST-API-Service mit verschiedenen Tools auf unterschiedliche Weise implementieren. Die beschriebene Lösung ist nicht die einzig richtige. Ich habe die Implementierung und die Tools basierend auf meinen persönlichen Erfahrungen und Vorlieben ausgewählt.

• Was werden wir machen?
• Welche Tools zur Auswahl?
• Entwicklung
  
  • Warum müssen Sie mit setup.py beginnen?
  • Wie werden Abhängigkeitsversionen angegeben?
  • Datenbank
    
    • Wir entwerfen das Schema
    • Beschreiben Sie das Schema in SQLAlchemy
    • Passen Sie Alembic an
    • Wir generieren Migrationen
  • Anwendung
    • Datenserialisierung
    • Handler
      • POST / Importe
      • GET / imports / $ import_id / Bürger
      • PATCH / imports / $ import_id / bürger / $ bürger_id
      • GET / imports / $ import_id / Bürger / Geburtstage
      • GET / imports / $ import_id / Towns / Stat / Perzentil / Alter
• Testen
  • Handler
    • GET / imports / $ import_id / Bürger
    • POST / Importe
    • PATCH / imports / $ import_id / bürger / $ bürger_id
    • GET / imports / $ import_id / Bürger / Geburtstage
    • GET / imports / $ import_id / Towns / Stat / Perzentil / Alter
  • Migrationen
• Versammlung
• Ci
• Bereitstellen
• Belastbarkeitstest
• Was kann man noch tun?
• Abschließend

Was werden wir machen?

• POST /imports
  Fügt einen neuen Upload mit Daten hinzu.
  
• GET /imports/$import_id/citizens
  Gibt die Bewohner der angegebenen Entlastung zurück;
  
• PATCH /imports/$import_id/citizens/$citizen_id
  Ändert Informationen über den Bewohner (und seine Verwandten) in der angegebenen Entladung;
  
• GET /imports/$import_id/citizens/birthdays
  , ( ), ;
  
• GET /imports/$import_id/towns/stat/percentile/age
  50-, 75- 99- ( ) .
  

?

Entwicklung

$ 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

Warum müssen Sie mit setup.py beginnen?

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 .

Wie werden Abhängigkeitsversionen angegeben?

• MAJOR - erhöht sich, wenn rückwärts inkompatible Änderungen hinzugefügt werden;
• MINOR - Erhöht sich beim Hinzufügen neuer Funktionen mit Unterstützung der Abwärtskompatibilität.
• PATCH - erhöht sich beim Hinzufügen von Fehlerkorrekturen mit Abwärtskompatibilitätsunterstützung.

Datenbank

Wir entwerfen das Schema

{
  "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. Die Importtabelle besteht aus einer automatisch inkrementierenden Spalte import_id. Es ist erforderlich, eine Fremdschlüsselprüfung in der Tabelle zu erstellen citizens.
   
2. In der Bürger- Tabelle werden skalare Daten zum Bewohner gespeichert (alle Felder außer Informationen zu familiären Beziehungen).
   
   Ein Paar ( import_id, citizen_id) wird als Primärschlüssel verwendet , um die Eindeutigkeit der Bewohner citizen_idinnerhalb des Rahmens zu gewährleisten import_id.
   
   Ein Fremdschlüssel citizens.import_id -> imports.import_idstellt sicher, dass das Feld citizens.import_idnur vorhandene Entladungen enthält.
   
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 .

Beschreiben Sie das 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')
    ),
)

Passen Sie Alembic an

1. Installationspaket: pip install alembic
2. Alembic initialisieren : cd analyzer && alembic init db/alembic.
   
   Dieser Befehl erstellt eine Konfigurationsdatei analyzer/alembic.iniund einen Ordner analyzer/db/alembicmit den folgenden Inhalten:
   
   • env.py- Wird jedes Mal angerufen, wenn Sie Alembic starten. Stellt eine Verbindung zur Alembic-Registrierung sqlalchemy.MetaDatamit einer Beschreibung des gewünschten Status der Datenbank her und enthält Anweisungen zum Starten von Migrationen.
     
   • script.py.mako - die Vorlage, auf deren Grundlage Migrationen generiert werden.
   • versions - Der Ordner, in dem Alembic Migrationen durchsucht (und generiert).
3. Geben Sie die Datenbankadresse in der Datei alembic.ini an:
   
   

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

4. Geben Sie eine Beschreibung des gewünschten Status der Datenbank (Registrierung sqlalchemy.MetaData) an, damit Alembic automatisch Migrationen generieren kann:
   
   

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

1. Das Dienstprogramm alembicsucht alembic.iniim aktuellen Arbeitsverzeichnis. Sie alembic.inikönnen den Pfad zum Befehlszeilenargument angeben, dies ist jedoch unpraktisch: Ich möchte den Befehl von jedem Ordner aus ohne zusätzliche Parameter aufrufen können.
2. Um Alembic für die Arbeit mit einer bestimmten Datenbank zu konfigurieren, müssen Sie die Datei ändern alembic.ini. Es wäre viel bequemer, beispielsweise die Datenbankeinstellungen für die Umgebungsvariable und / oder ein Befehlszeilenargument anzugeben --pg-url.
3. Der Name des Dienstprogramms alembickorreliert nicht sehr gut mit dem Namen unseres Dienstes (und der Benutzer hat möglicherweise überhaupt kein Python und weiß nichts über Alembic). Für den Endbenutzer wäre es viel bequemer, wenn beispielsweise alle ausführbaren Befehle des Dienstes ein gemeinsames Präfix hätten analyzer-*.

• Alembic verwendet ein Standardmodul zur Verarbeitung von Befehlszeilenargumenten argparse. Sie können ein optionales Argument --pg-urlmit einem Standardwert aus einer Umgebungsvariablen hinzufügen ANALYZER_PG_URL.
  
  
  Der 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()

• Der Pfad zur Datei alembic.inikann relativ zum Speicherort der ausführbaren Datei und nicht zum aktuellen Arbeitsverzeichnis des Benutzers berechnet werden.
  
  
  Der 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]'

Wir generieren Migrationen

$ 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

• Die in den erstellten Tabellen angegebenen Benutzerdatentypen werden automatisch erstellt (in unserem Fall - gender), aber der Code zum Löschen wird downgradenicht generiert. Wenn Sie die Migration anwenden, zurücksetzen und dann erneut anwenden, führt dies zu einem Fehler, da der angegebene Datentyp bereits vorhanden ist.
  
  
  Löschen Sie den Geschlechtsdatentyp in der Downgrade-Methode

  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 der Methode können downgradeeinige Aktionen manchmal entfernt werden (wenn wir die gesamte Tabelle löschen, können Sie ihre Indizes nicht separat löschen):
  
  
  zum Beispiel

  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

Anwendung

import logging

from aiohttp import web


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

    #  
    app = web.Application()

    #  
    app.router.add_route(...)

    #  
    web.run_app(app)

• Wie konfiguriere ich die Anwendung? Sie müssen mindestens den Host und den Port für die Verbindung mit Clients sowie Informationen für die Verbindung mit der Datenbank angeben.
  
  Ich möchte dieses Problem wirklich gerne mit Hilfe des Moduls lösen ConfigArgParse: Es erweitert das Standardmodul argparseund ermöglicht die Verwendung von Befehlszeilenargumenten, Umgebungsvariablen (unverzichtbar für die Konfiguration von Docker-Containern) und sogar Konfigurationsdateien (sowie die Kombination dieser Methoden) für die Konfiguration. Mit ConfigArgParsedieser Funktion können Sie auch die Werte der Anwendungskonfigurationsparameter überprüfen.
  
  
  Ein Beispiel für die Verarbeitung von Parametern mit 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)

Handler

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

• Datentyp, Format und Verfügbarkeit aller erforderlichen Felder;
• Mangel an unbekannten Feldern;
• Das Geburtsdatum muss im Format angegeben werden DD.MM.YYYYund kann für die Zukunft keine Bedeutung haben.
• Die Liste der Verwandten jedes Bewohners muss eindeutige Kennungen der in diesem Upload vorhandenen Bewohner enthalten.

• citizen_id Jeder Bewohner des Entladens sollte eindeutig sein.
• Familienbande sollten wechselseitig sein (wenn Bewohner Nr. 1 einen Bewohner Nr. 2 in der Liste der Verwandten hat, muss Bewohner Nr. 2 auch einen Verwandten Nr. 1 haben).

GET / imports / $ import_id / Bürger

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. Ein SelectQueryTyp - Objekt , AsyncIterabledas Datensätze aus der Datenbank zurückgibt. Beim ersten Aufruf wird eine Verbindung zur Datenbank hergestellt, eine Transaktion geöffnet und ein Cursor erstellt. Während der weiteren Iteration werden Datensätze aus der Datenbank zurückgegeben. Es wird vom Handler zurückgegeben.
   
   
   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. Ein Serializer AsyncGenJSONListPayload, der über asynchrone Generatoren iterieren, Daten von einem asynchronen Generator an JSON serialisieren und Daten in Teilen an Clients senden kann. Es ist aiohttp.PAYLOAD_REGISTRYals Serializer von Objekten registriert 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 / bürger / $ bürger_id

• Art und Format der Daten für die angegebenen Felder.
• Geburtsdatum. Es muss in einem Format angegeben werden DD.MM.YYYYund kann für die Zukunft nicht von Bedeutung sein.
• Eine Liste der Verwandten jedes Bewohners. Es muss eindeutige Kennungen für die Bewohner haben.

GET / imports / $ import_id / Bürger / Geburtstage

1. Rufen Sie Daten für Einwohner mit Verwandten aus der Datenbank ab und aggregieren Sie auf Python-Seite Daten nach Monat und generieren Sie Listen für die Monate, für die keine Daten in der Datenbank vorhanden sind.
2. Kompilieren Sie eine JSON-Anfrage in der Datenbank und fügen Sie Stubs für die fehlenden Monate hinzu.

GET / imports / $ import_id / Towns / Stat / Perzentil / Alter

1. Holen Sie sich das Alter der Einwohner aus der Datenbank, gruppiert nach Stadt, und berechnen Sie dann auf der Python-Seite die Perzentile mit numpy (das in der Aufgabe als Referenz angegeben wird) und runden Sie auf zwei Dezimalstellen auf.
2. PostgreSQL: percentile_cont , SQL-, numpy .

Testen

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)

Handler

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 / Bürger

• Entladen mit mehreren Verwandten. Überprüft, ob für jeden Einwohner eine Liste mit Kennungen von Verwandten korrekt erstellt wird.
• Entladen mit einem Bewohner ohne Verwandte. Überprüft, ob das Feld relativeseine leere Liste ist (aufgrund LEFT JOINder SQL-Abfrage kann die Liste der Verwandten gleich sein [None]).
• Entladen mit einem Bewohner, der ein Verwandter von sich selbst ist.
• Leeres Entladen. Überprüft, ob der Handler das Hinzufügen eines leeren Entladens zulässt und nicht mit einem Fehler abstürzt.

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

• Richtige Daten, die voraussichtlich erfolgreich zur Datenbank hinzugefügt werden.
  
  
  • ( ).
    
    . , , insert , .
  • ( , ).
    
    , .
    
  • .
    
    , . :)
    
  • 
    
    , aiohttp PostgreSQL 32 767 ( ).
    
  • Leeres Entladen
    
    Der Handler sollte einen solchen Fall berücksichtigen und nicht fallen und versuchen, mit den Bewohnern eine leere Einfügung in den Tisch durchzuführen.
    
  
  
• Bei fehlerhaften Daten wird eine HTTP-Antwort von 400: Bad Request erwartet.
  
  
  • Geburtsdatum ist falsch (Zukunftsform).
  • bürger_id ist innerhalb des Uploads nicht eindeutig.
  • Eine Verwandtschaft wird falsch angegeben (es gibt nur von einem Bewohner zum anderen, aber es gibt keine Rückmeldung).
  • Der Bewohner hat einen nicht existierenden Verwandten beim Entladen.
  • Familienbande sind nicht einzigartig.

PATCH / imports / $ import_id / bürger / $ bürger_id

• Das Feld ist angegeben, hat jedoch einen falschen Datentyp und / oder ein falsches Format
• Das Geburtsdatum ist falsch (zukünftige Zeit).
• Das Feld relativesenthält einen Verwandten, der beim Entladen nicht vorhanden ist.

GET / imports / $ import_id / Bürger / Geburtstage

• Eine Entladung, bei der ein Bewohner in einem Monat einen Verwandten und in einem anderen zwei Verwandte hat.
• Entladen mit einem Bewohner ohne Verwandte. Überprüft, ob der Handler dies bei den Berechnungen nicht berücksichtigt.
• Leeres Entladen. Überprüft, ob der Handler nicht fehlschlägt und das richtige Wörterbuch mit einer Antwort von 12 Monaten zurückgibt.
• Entladen mit einem Bewohner, der ein Verwandter von sich selbst ist. Überprüft, ob ein Bewohner ein Geschenk für den Monat seiner Geburt kauft.

GET / imports / $ import_id / Towns / Stat / Perzentil / Alter

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

• Entladen mit mehreren Bewohnern, deren Geburtstag morgen ist (Alter - mehrere Jahre und 364 Tage). Überprüft, ob der Prozessor nur die Anzahl der vollen Jahre für die Berechnungen verwendet.
• Entladen mit einem Bewohner, dessen Geburtstag heute ist (Alter - genau einige Jahre). Es prüft den regionalen Fall - das Alter eines Bewohners, dessen Geburtstag heute ist, sollte nicht als um 1 Jahr verkürzt berechnet werden.
• Leeres Entladen. Der Handler sollte nicht darauf fallen.

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

Eine interessante Tatsache: Als ich diesen Test am 29. Februar 2020 schrieb, hörte ich aufgrund eines Fehlers in Faker plötzlich auf, Entladungen mit Anwohnern zu generieren (2020 ist ein Schaltjahr, und andere Jahre, die Faker gewählt hatte, waren nicht immer auch Schaltjahre in ihnen war nicht der 29. Februar). Denken Sie daran, Daten aufzuzeichnen und Randfälle zu testen!

Migrationen

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

Versammlung

1. snakepacker/python:allErstellen Sie mit einem „schweren“ Image (~ 1 GB, ~ 500 MB komprimiert) eine virtuelle Umgebung, installieren Sie alle Abhängigkeiten und das Anwendungspaket darin. Dieses Image wird ausschließlich für die Assembly benötigt. Es kann einen Compiler, alle erforderlichen Bibliotheken und Dateien mit Headern enthalten.
   
   

   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. Wir kopieren die fertige virtuelle Umgebung in ein "leichtes" Image snakepacker/python:3.8(~ 100 MB, komprimiert ~ 50 MB), das nur den Interpreter der erforderlichen Version von Python enthält.
   
   Wichtig: In einer virtuellen Umgebung werden absolute Pfade verwendet, daher muss sie an dieselbe Adresse kopiert werden, an der sie im Kollektorcontainer zusammengestellt wurde.
   
   

   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

Bereitstellen

[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

Über Stresstests

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

Was kann man noch tun?

* * *

Danksagung