يمكنك تنفيذ خدمة REST API بطرق مختلفة باستخدام أدوات مختلفة. الحل الموصوف ليس هو الحل الوحيد الصحيح ، لقد اخترت التنفيذ والأدوات بناءً على تجربتي الشخصية وتفضيلاتي.

• ماذا نفعل؟
• ما الأدوات للاختيار؟
• تطوير
  
  • لماذا تحتاج إلى البدء بـ setup.py؟
  • كيفية تحديد إصدارات التبعية؟
  • قاعدة البيانات
    
    • نقوم بتصميم المخطط
    • وصف المخطط في SQLAlchemy
    • تخصيص Alembic
    • نحن نولد الهجرات
  • تطبيق
    • تسلسل البيانات
    • معالجات
      • ما بعد / الواردات
      • الحصول على / الواردات / $ import_id / المواطنين
      • التصحيح / الواردات / $ import_id / للمواطنين / $ مواطن_المعروف
      • الحصول على / الواردات / $ import_id / المواطنين / أعياد الميلاد
      • الحصول على / الواردات / $ import_id / المدن / ستات / النسبة المئوية / العمر
• اختبارات
  • معالجات
    • الحصول على / الواردات / $ import_id / المواطنين
    • ما بعد / الواردات
    • التصحيح / الواردات / $ import_id / للمواطنين / $ مواطن_المعروف
    • الحصول على / الواردات / $ import_id / المواطنين / أعياد الميلاد
    • الحصول على / الواردات / $ import_id / المدن / ستات / النسبة المئوية / العمر
  • الهجرات
• المجسم
• Ci
• نشر
• اختبار الإجهاد
• ما الذي يمكن فعله أيضًا؟
• أخيرا

ماذا نفعل؟

• 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
  50-, 75- 99- ( ) .
  

?

تطوير

$ 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

لماذا تحتاج إلى البدء بـ 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 .

كيفية تحديد إصدارات التبعية؟

• MAJOR - يزيد عند إضافة تغييرات غير متوافقة مع الإصدارات السابقة ؛
• MINOR - زيادة عند إضافة وظائف جديدة مع دعم التوافق العكسي ؛
• PATCH - يزيد عند إضافة إصلاحات الأخطاء مع دعم التوافق مع الإصدارات السابقة.

قاعدة البيانات

نقوم بتصميم المخطط

{
  "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. يتكون جدول الاستيراد من عمود متزايد تلقائيًا import_id. هناك حاجة لإنشاء تدقيق مفتاح خارجي في الجدول citizens.
   
2. يقوم جدول المواطنين بتخزين البيانات العددية عن المقيم (جميع المجالات باستثناء المعلومات حول العلاقات الأسرية).
   
   يتم استخدام الزوج ( import_id، citizen_id) كمفتاح أساسي ، مما يضمن تفرد المقيمين citizen_idداخل الإطار import_id. يضمن
   
   المفتاح الخارجي citizens.import_id -> imports.import_idاحتواء الحقل citizens.import_idعلى عمليات إلغاء التحميل الموجودة فقط.
   
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 .

وصف المخطط في 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')
    ),
)

تخصيص Alembic

1. ثبت المجموعة: pip install alembic
2. تهيئة الإنبيق: cd analyzer && alembic init db/alembic.
   
   سيقوم هذا الأمر بإنشاء ملف تكوين analyzer/alembic.iniومجلد analyzer/db/alembicبالمحتويات التالية:
   
   • env.py- يسمى في كل مرة تبدأ فيها Alembic. يتصل بتسجيل Alembic sqlalchemy.MetaDataمع وصف الحالة المرغوبة لقاعدة البيانات ويحتوي على إرشادات لبدء عمليات الترحيل.
     
   • script.py.mako - القالب الذي يتم على أساسه إنشاء عمليات الترحيل.
   • versions - المجلد الذي ستبحث فيه Alembic (وتولد) عمليات الترحيل.
3. حدد عنوان قاعدة البيانات في ملف alembic.ini:
   
   

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

4. حدد وصفًا للحالة المطلوبة لقاعدة البيانات (التسجيل sqlalchemy.MetaData) بحيث يمكن لـ Alembic إنشاء عمليات الترحيل تلقائيًا:
   
   

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

1. تقوم الأداة المساعدة alembicبالبحث alembic.iniفي دليل العمل الحالي. يمكنك alembic.iniتحديد المسار إلى وسيطة سطر الأوامر ، ولكن هذا غير مريح: أريد أن أكون قادرًا على استدعاء الأمر من أي مجلد بدون معلمات إضافية.
2. لتكوين Alembic للعمل مع قاعدة بيانات محددة ، تحتاج إلى تغيير الملف alembic.ini. سيكون من الأنسب تحديد إعدادات قاعدة البيانات لمتغير البيئة و / أو وسيطة سطر الأوامر ، على سبيل المثال --pg-url.
3. alembicلا يرتبط اسم الأداة بشكل جيد مع اسم خدمتنا (وقد لا يكون لدى المستخدم Python على الإطلاق ولا يعرف شيئًا عن Alembic). سيكون أكثر ملاءمة للمستخدم النهائي إذا كانت جميع الأوامر القابلة للتنفيذ للخدمة لها بادئة مشتركة ، على سبيل المثال analyzer-*.

• تستخدم Alembic وحدة نمطية قياسية لمعالجة وسيطات سطر الأوامر argparse. يسمح لك بإضافة وسيطة اختيارية --pg-urlبقيمة افتراضية من متغير بيئة ANALYZER_PG_URL.
  
  
  الرمز

  import os
  from alembic.config import CommandLine, Config
  from analyzer.utils.pg import DEFAULT_PG_URL
  
  
  def main():
      alembic = CommandLine()
      alembic.parser.add_argument(
          '--pg-url', default=os.getenv('ANALYZER_PG_URL', DEFAULT_PG_URL),
          help='Database URL [env var: ANALYZER_PG_URL]'
      )
      options = alembic.parser.parse_args()
  
      #    Alembic
      config = Config(file_=options.config, ini_section=options.name,
                      cmd_opts=options)
  
      #   sqlalchemy.url   Alembic
      config.set_main_option('sqlalchemy.url', options.pg_url)
  
      #   alembic
      exit(alembic.run_cmd(config, options))
  
  
  if __name__ == '__main__':
      main()

• alembic.iniيمكن حساب مسار الملف بالنسبة لموقع الملف القابل للتنفيذ ، وليس دليل العمل الحالي للمستخدم.
  
  
  الرمز

  import os
  from alembic.config import CommandLine, Config
  from pathlib import Path
  
  
  PROJECT_PATH = Path(__file__).parent.parent.resolve()
  
  
  def main():
      alembic = CommandLine()
      options = alembic.parser.parse_args()
  
      #     (alembic.ini),   
      #    
      if not os.path.isabs(options.config):
          options.config = os.path.join(PROJECT_PATH, options.config)
  
      #    Alembic
      config = Config(file_=options.config, ini_section=options.name,
                      cmd_opts=options)
  
      #      alembic   (,  alembic
      #   env.py,       )
      alembic_location = config.get_main_option('script_location')
      if not os.path.isabs(alembic_location):
          config.set_main_option('script_location',
                                 os.path.join(PROJECT_PATH, alembic_location))
  
      #   alembic
      exit(alembic.run_cmd(config, options))
  
  
  if __name__ == '__main__':
      main()

from setuptools import setup

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

$ pip install -e '.[dev]'

نحن نولد الهجرات

$ 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

• يتم إنشاء أنواع بيانات المستخدم المحددة في الجداول التي تم إنشاؤها تلقائيًا (في حالتنا - gender) ، ولكن downgradeلم يتم إنشاء الرمز لحذفها . إذا قمت بتطبيق التراجع ، ثم التراجع ، ثم تطبيق الترحيل مرة أخرى ، سيؤدي ذلك إلى حدوث خطأ لأن نوع البيانات المحدد موجود بالفعل.
  
  
  احذف نوع بيانات الجنس في طريقة الرجوع إلى إصدار أقدم

  from alembic import op
  from sqlalchemy import Column, Enum
  
  GenderType = Enum('female', 'male', name='gender')
  
  
  def upgrade():
      ...
      #      GenderType   
      op.create_table('citizens', ...,
                      Column('gender', GenderType, nullable=False))
      ...
  
  
  def downgrade():
      op.drop_table('citizens')
  
      #       
      GenderType.drop(op.get_bind())

• في الطريقة ، downgradeيمكن في بعض الأحيان إزالة بعض الإجراءات (إذا حذفنا الجدول بأكمله ، لا يمكنك حذف فهارسه بشكل منفصل):
  
  
  على سبيل المثال

  def downgrade():
  op.drop_table('relations')
  
  #      citizens,    
  #    
  op.drop_index(op.f('ix__citizens__town'), table_name='citizens')
  op.drop_table('citizens')
  op.drop_table('imports')

$ analyzer-db upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> d5f704ed4610, Initial

تطبيق

import logging

from aiohttp import web


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

    #  
    app = web.Application()

    #  
    app.router.add_route(...)

    #  
    web.run_app(app)

• كيفية تكوين التطبيق؟ كحد أدنى ، يجب عليك تحديد المضيف والمنفذ لتوصيل العملاء ، بالإضافة إلى معلومات الاتصال بقاعدة البيانات.
  
  أحب حقًا حل هذه المشكلة بمساعدة الوحدة النمطية ConfigArgParse: فهي تمد المشكلة القياسية argparseوتسمح باستخدام وسيطات سطر الأوامر ومتغيرات البيئة (التي لا غنى عنها لتكوين حاويات Docker) وحتى ملفات التكوين (بالإضافة إلى دمج هذه الأساليب) للتكوين. باستخدامه ConfigArgParse، يمكنك أيضًا التحقق من قيم معلمات تكوين التطبيق.
  
  
  مثال على معلمات المعالجة باستخدام ConfigArgParse

  from aiohttp import web
  from configargparse import ArgumentParser, ArgumentDefaultsHelpFormatter
  
  from analyzer.utils.argparse import positive_int
  
  parser = ArgumentParser(
      #        ANALYZER_,
      #  ANALYZER_API_ADDRESS  ANALYZER_API_PORT
      auto_env_var_prefix='ANALYZER_',
  
      #     
      formatter_class=ArgumentDefaultsHelpFormatter
  )
  
  parser.add_argument('--api-address', default='0.0.0.0',
                      help='IPv4/IPv6 address API server would listen on')
  
  #      
  parser.add_argument('--api-port', type=positive_int, default=8081,
                      help='TCP port API server would listen on')
  
  
  def main():
      #   ,     
      #  ,    
      args = parser.parse_args()
  
      #       
      app = web.Application()
      web.run_app(app, host=args.api_address, port=args.api_port)
  
  
  if __name__ == '__main__':
      main()

  
  , ConfigArgParse, argparse, ( -h --help). :
  
  

  $ python __main__.py --help
  usage: __main__.py [-h] [--api-address API_ADDRESS] [--api-port API_PORT]
  
  If an arg is specified in more than one place, then commandline values override environment variables which override defaults.
  
  optional arguments:
    -h, --help            show this help message and exit
    --api-address API_ADDRESS
                          IPv4/IPv6 address API server would listen on [env var: ANALYZER_API_ADDRESS] (default: 0.0.0.0)
    --api-port API_PORT   TCP port API server would listen on [env var: ANALYZER_API_PORT] (default: 8081)

• — , «» . , .
  
  os.environ.clear(), Python (, asyncio?), , ConfigArgParser.
  
  

  import os
  from typing import Callable
  from configargparse import ArgumentParser
  from yarl import URL
  
  from analyzer.api.app import create_app
  from analyzer.utils.pg import DEFAULT_PG_URL
  
  ENV_VAR_PREFIX = 'ANALYZER_'
  
  parser = ArgumentParser(auto_env_var_prefix=ENV_VAR_PREFIX)
  parser.add_argument('--pg-url', type=URL, default=URL(DEFAULT_PG_URL),
                     help='URL to use to connect to the database')
  
  
  def clear_environ(rule: Callable):
      """
        ,     
       rule
      """
      #   os.environ    tuple,    
      # os.environ   
      for name in filter(rule, tuple(os.environ)):
          os.environ.pop(name)
  
  
  def main():
      #  
      args = parser.parse_args()
  
      #      ANALYZER_
      clear_environ(lambda i: i.startswith(ENV_VAR_PREFIX))
  
      #  
      app = create_app(args)
      ...
  
  
  if __name__ == '__main__':
      main()

• stderr/ .
  
  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)

معالجات

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']

ما بعد / الواردات

• نوع البيانات وتنسيق وتوافر جميع الحقول المطلوبة ؛
• عدم وجود حقول غير مألوفة ؛
• يجب تحديد تاريخ الميلاد بالصيغة DD.MM.YYYYولا يمكن أن يكون له أي أهمية من المستقبل ؛
• يجب أن تحتوي قائمة أقارب كل مقيم على معرفات فريدة للمقيمين الموجودين في هذا التحميل.

• citizen_id يجب أن يكون كل مقيم في التفريغ فريدًا ؛
• يجب أن تكون الروابط العائلية ثنائية الاتجاه (إذا كان المقيم رقم 1 لديه مقيم رقم 2 في قائمة الأقارب ، فيجب أن يكون للمقيم رقم 2 أيضًا قريب رقم 1).

الحصول على / الواردات / $ import_id / المواطنين

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. عنصر SelectQueryنوع يقوم AsyncIterableبارجاع السجلات من قاعدة البيانات. في المكالمة الأولى ، يتصل بقاعدة البيانات ، ويفتح المعاملات وينشئ المؤشر ؛ وأثناء التكرار الإضافي ، يقوم بإرجاع السجلات من قاعدة البيانات. يتم إرجاعها بواسطة المعالج.
   
   
   حدد رمز الاستعلام

   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. مُسلسل AsyncGenJSONListPayloadيمكنه التكرار عبر المولدات غير المتزامنة ، وتسلسل البيانات من المولد غير المتزامن إلى JSON وإرسال البيانات إلى العملاء في أجزاء. يتم تسجيله aiohttp.PAYLOAD_REGISTRYكمسلسل للأشياء AsyncIterable.
   
   
   كود 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)

التصحيح / الواردات / $ import_id / للمواطنين / $ مواطن_المعروف

• نوع وتنسيق البيانات للحقول المحددة.
• تاريخ الولادة. يجب تحديده بتنسيق DD.MM.YYYYولا يمكن أن يكون ذا أهمية من المستقبل.
• قائمة أقارب كل مقيم. يجب أن يكون لها معرفات فريدة للمقيمين.

الحصول على / الواردات / $ import_id / المواطنين / أعياد الميلاد

1. احصل على بيانات للمقيمين مع الأقارب من قاعدة البيانات ، وعلى جانب Python ، قم بتجميع البيانات حسب الشهر وإنشاء قوائم لتلك الأشهر التي لا توجد بيانات لها في قاعدة البيانات.
2. تجميع طلب json في قاعدة البيانات وإضافة بذرة للأشهر المفقودة.

الحصول على / الواردات / $ import_id / المدن / ستات / النسبة المئوية / العمر

1. احصل على عمر السكان من قاعدة البيانات ، مجمعة حسب المدينة ، ثم على جانب Python ، احسب النسب المئوية باستخدام numpy (الذي تم تحديده كمرجع في المهمة) وقرب ما يصل إلى منزلين عشريين.
2. PostgreSQL: percentile_cont , SQL-, numpy .

اختبارات

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)

معالجات

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 .

الحصول على / الواردات / $ import_id / المواطنين

• التفريغ مع عدة أقارب. يتحقق من أنه سيتم تشكيل قائمة لكل مقيم مع معرفات الأقارب بشكل صحيح.
• التفريغ مع مقيم واحد بدون أقارب. للتحقق من أن الحقل relativesعبارة عن قائمة فارغة (نظرًا LEFT JOINلاستعلام SQL ، قد تكون قائمة الأقارب متساوية [None]).
• التفريغ مع مقيم من أقاربه.
• تفريغ فارغ. التحقق من أن المعالج يسمح بإضافة تفريغ فارغ ولا يتلف مع وجود خطأ.

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

ما بعد / الواردات

• البيانات الصحيحة ، المتوقع إضافتها بنجاح إلى قاعدة البيانات.
  
  
  • ( ).
    
    . , , insert , .
  • ( , ).
    
    , .
    
  • .
    
    , . :)
    
  • 
    
    , aiohttp PostgreSQL 32 767 ( ).
    
  • تفريغ فارغ
    
    يجب على المعالج أن يأخذ في الاعتبار مثل هذه الحالة وألا يسقط ، محاولاً إجراء إدراج فارغ في الجدول مع السكان.
    
  
  
• البيانات التي بها أخطاء ، توقع استجابة HTTP 400: طلب غير صحيح.
  
  
  • تاريخ الميلاد غير صحيح (زمن المستقبل).
  • سيتيزنيد ليست فريدة في التحميل.
  • يشار إلى القرابة بشكل غير صحيح (لا يوجد سوى من مقيم لآخر ، ولكن لا توجد تعليقات).
  • المقيم لديه قريب غير موجود في التفريغ.
  • العلاقات الأسرية ليست فريدة من نوعها.

التصحيح / الواردات / $ import_id / للمواطنين / $ مواطن_المعروف

• تم تحديد الحقل ، ولكن به نوع بيانات و / أو تنسيق غير صحيح
• تاريخ الميلاد غير صحيح (في المستقبل).
• relativesيحتوي الحقل على قريب غير موجود في التفريغ.

الحصول على / الواردات / $ import_id / المواطنين / أعياد الميلاد

• تفريغ يكون للمقيم فيه قريب في شهر واحد واثنين من أقاربه في آخر.
• التفريغ مع مقيم واحد بدون أقارب. التحقق من أن المعالج لا يأخذها في الاعتبار في الحسابات.
• تفريغ فارغ. التحقق من أن المعالج لن يفشل وسيعيد القاموس الصحيح خلال 12 شهرًا من الاستجابة.
• التفريغ مع مقيم من أقاربه. الشيكات أن المقيم سيشتري هدية لشهر ولادته.

الحصول على / الواردات / $ import_id / المدن / ستات / النسبة المئوية / العمر

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 يومًا). يتحقق من أن المعالج يستخدم عدد السنوات الكاملة فقط في الحسابات.
• التفريغ مع مقيم عيد ميلاده اليوم (العمر - بضع سنوات بالضبط). ويتحقق من الحالة الإقليمية - يجب ألا يحسب عمر المقيم الذي يكون عيد ميلاده اليوم مخفضًا بمقدار عام واحد.
• تفريغ فارغ. يجب ألا يقع المعالج عليه.

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

حقيقة مثيرة للاهتمام: عندما كتبت هذا الاختبار في 29 فبراير 2020 ، توقفت فجأة عن توليد حمولات مع المقيمين بسبب خطأ في Faker (2020 هي سنة كبيسة ، والسنوات الأخرى التي اختارها Faker لم تكن دائمًا قفزة فيها أيضًا لم يكن 29 فبراير). تذكر تسجيل التواريخ وحالات حافة الاختبار!

الهجرات

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

المجسم

1. باستخدام صورة "ثقيلة" snakepacker/python:all(~ 1 جيجا بايت ، ~ 500 ميجا بايت مضغوطة) ، قم بإنشاء بيئة افتراضية ، وقم بتثبيت كل التبعيات وحزمة التطبيق فيها. هذه الصورة مطلوبة حصريًا للتجميع ، ويمكن أن تحتوي على مترجم ، وجميع المكتبات والملفات الضرورية ذات الرؤوس.
   
   

   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. نقوم بنسخ البيئة الافتراضية النهائية إلى صورة "خفيفة" snakepacker/python:3.8(~ 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"]

###############      ################
#  — «» (~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

نشر

[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

حول اختبار الإجهاد

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

ما الذي يمكن فعله أيضًا؟

* * *

شكر وتقدير