Anda dapat mengimplementasikan layanan REST API dengan berbagai cara menggunakan alat yang berbeda. Solusi yang dijelaskan bukan satu-satunya yang tepat, saya memilih implementasi dan alat berdasarkan pengalaman dan preferensi pribadi saya.

• Apa yang kita lakukan?
• Alat apa yang harus dipilih?
• Pengembangan
  
  • Mengapa Anda harus mulai dengan setup.py?
  • Bagaimana cara menentukan versi ketergantungan?
  • Basis data
    
    • Kami merancang skema
    • Jelaskan skema dalam SQLAlchemy
    • Sesuaikan Alembic
    • Kami menghasilkan migrasi
  • aplikasi
    • Serialisasi data
    • Penangan
      • POST / impor
      • DAPATKAN / impor / $ import_id / warga
      • PATCH / impor / $ import_id / warga negara / $ citizen_id
      • DAPATKAN / impor / $ import_id / warga negara / ulang tahun
      • DAPATKAN / impor / $ import_id / kota / stat / persentil / usia
• Pengujian
  • Penangan
    • DAPATKAN / impor / $ import_id / warga
    • POST / impor
    • PATCH / impor / $ import_id / warga negara / $ citizen_id
    • DAPATKAN / impor / $ import_id / warga negara / ulang tahun
    • DAPATKAN / impor / $ import_id / kota / stat / persentil / usia
  • Migrasi
• Majelis
• Ci
• Menyebarkan
• Pengujian Stres
• Apa lagi yang bisa dilakukan?
• Akhirnya

Apa yang kita lakukan?

• POST /imports
  Menambahkan unggahan baru dengan data;
  
• GET /imports/$import_id/citizens
  Mengembalikan penghuni tempat pembuangan yang ditentukan;
  
• PATCH /imports/$import_id/citizens/$citizen_id
  Mengubah informasi tentang penduduk (dan kerabatnya) dalam bongkar yang ditentukan;
  
• GET /imports/$import_id/citizens/birthdays
  , ( ), ;
  
• GET /imports/$import_id/towns/stat/percentile/age
  50-, 75- 99- ( ) .
  

?

Pengembangan

$ 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

Mengapa Anda harus mulai dengan 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 .

Bagaimana cara menentukan versi ketergantungan?

• MAJOR - meningkat ketika perubahan yang tidak kompatibel mundur ditambahkan;
• MINOR - Meningkat ketika menambahkan fungsionalitas baru dengan dukungan untuk kompatibilitas mundur;
• PATCH - meningkat ketika menambahkan perbaikan bug dengan dukungan kompatibilitas mundur.

Basis data

Kami merancang skema

{
  "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. Tabel impor terdiri dari kolom yang bertambah secara otomatis import_id. Diperlukan untuk membuat pemeriksaan kunci asing di tabel citizens.
   
2. Tabel warga menyimpan data skalar tentang penduduk (semua bidang kecuali informasi tentang hubungan keluarga).
   
   Sepasang ( import_id, citizen_id) digunakan sebagai kunci utama , menjamin keunikan penghuni citizen_iddalam kerangka kerja import_id.
   
   Kunci asing citizens.import_id -> imports.import_idmemastikan bahwa bidang citizens.import_idhanya berisi pembongkaran yang ada.
   
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 .

Jelaskan skema dalam 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')
    ),
)

Sesuaikan Alembic

1. Instal Paket: pip install alembic
2. Initialize Alembic: cd analyzer && alembic init db/alembic.
   
   Perintah ini akan membuat file konfigurasi analyzer/alembic.inidan folder analyzer/db/alembicdengan konten berikut:
   
   • env.py- Dipanggil setiap kali Anda memulai Alembic. Terhubung ke registri Alembic sqlalchemy.MetaDatadengan deskripsi kondisi database yang diinginkan dan berisi instruksi untuk memulai migrasi.
     
   • script.py.mako - templat berdasarkan migrasi yang dihasilkan.
   • versions - folder tempat Alembic akan mencari (dan menghasilkan) migrasi.
3. Tentukan alamat database dalam file alembic.ini:
   
   

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

4. Tentukan deskripsi kondisi database (registri sqlalchemy.MetaData) yang diinginkan sehingga Alembic dapat menghasilkan migrasi secara otomatis:
   
   

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

1. Utilitas alembicmencari alembic.inidi direktori kerja saat ini. Anda alembic.inidapat menentukan jalur ke argumen baris perintah, tetapi ini tidak nyaman: Saya ingin dapat memanggil perintah dari folder apa pun tanpa parameter tambahan.
2. Untuk mengonfigurasi Alembic agar berfungsi dengan database tertentu, Anda perlu mengubah file alembic.ini. Akan jauh lebih mudah untuk menentukan pengaturan basis data untuk variabel lingkungan dan / atau argumen baris perintah, misalnya --pg-url.
3. Nama utilitas alembictidak berkorelasi sangat baik dengan nama layanan kami (dan pengguna mungkin sebenarnya tidak memiliki Python sama sekali dan tidak tahu apa-apa tentang Alembic). Akan jauh lebih nyaman bagi pengguna akhir jika semua perintah layanan yang dapat dieksekusi memiliki awalan umum, misalnya analyzer-*.

• Alembic menggunakan modul standar untuk memproses argumen baris perintah argparse. Ini memungkinkan Anda untuk menambahkan argumen opsional --pg-urldengan nilai default dari variabel lingkungan ANALYZER_PG_URL.
  
  
  Kode

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

• Jalur ke file alembic.inidapat dihitung relatif terhadap lokasi file yang dapat dieksekusi, dan bukan direktori kerja pengguna saat ini.
  
  
  Kode

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

Kami menghasilkan migrasi

$ 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

• Tipe data pengguna yang ditentukan dalam tabel yang dibuat dibuat secara otomatis (dalam kasus kami - gender), tetapi kode untuk menghapusnya downgradetidak dihasilkan. Jika Anda menerapkan, memutar kembali, dan kemudian menerapkan migrasi lagi, ini akan menyebabkan kesalahan karena tipe data yang ditentukan sudah ada.
  
  
  Hapus tipe data gender dalam metode 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())

• Dalam metode ini, downgradebeberapa tindakan terkadang dapat dihapus (jika kami menghapus seluruh tabel, Anda tidak dapat menghapus indeksnya secara terpisah):
  
  
  contohnya

  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

aplikasi

import logging

from aiohttp import web


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

    #  
    app = web.Application()

    #  
    app.router.add_route(...)

    #  
    web.run_app(app)

• Bagaimana cara mengkonfigurasi aplikasi? Minimal, Anda harus menentukan host dan port untuk menghubungkan klien, serta informasi untuk menghubungkan ke database.
  
  Saya benar-benar ingin menyelesaikan masalah ini dengan bantuan modul ConfigArgParse: ia memperluas yang standar argparsedan memungkinkan menggunakan argumen baris perintah, variabel lingkungan (sangat diperlukan untuk mengkonfigurasi wadah Docker) dan bahkan file konfigurasi (serta menggabungkan metode ini) untuk konfigurasi. Dengan menggunakannya ConfigArgParse, Anda juga dapat memvalidasi nilai parameter konfigurasi aplikasi.
  
  
  Contoh pemrosesan parameter menggunakan 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)

Penangan

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

• Jenis data, format, dan ketersediaan semua bidang yang diperlukan;
• Kurangnya bidang yang tidak dikenal;
• Tanggal lahir harus ditunjukkan dalam format DD.MM.YYYYdan tidak boleh penting dari masa depan;
• Daftar kerabat masing-masing penduduk harus berisi pengidentifikasi unik dari penduduk yang ada dalam unggahan ini.

• citizen_id setiap penduduk dalam bongkar muat harus unik;
• Ikatan keluarga harus dua arah (jika residen # 1 memiliki residen # 2 dalam daftar kerabat, maka residen # 2 juga harus memiliki kerabat # 1).

DAPATKAN / impor / $ import_id / warga

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. SelectQueryTipe objek AsyncIterableyang mengembalikan catatan dari database. Pada panggilan pertama, terhubung ke database, membuka transaksi dan membuat kursor, selama iterasi lebih lanjut, ia mengembalikan catatan dari database. Itu dikembalikan oleh pawang.
   
   
   Pilih Kode Kueri

   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. Serializer AsyncGenJSONListPayloadyang dapat beralih melalui generator asinkron, membuat serialisasi data dari generator asinkron ke JSON dan mengirim data ke klien di beberapa bagian. Terdaftar aiohttp.PAYLOAD_REGISTRYsebagai serializer objek AsyncIterable.
   
   
   Kode 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 / impor / $ import_id / warga negara / $ citizen_id

• Jenis dan format data untuk bidang yang ditentukan.
• Tanggal lahir. Itu harus ditentukan dalam format DD.MM.YYYYdan tidak bisa penting dari masa depan.
• Daftar kerabat masing-masing penduduk. Itu harus memiliki pengidentifikasi unik untuk penduduk.

DAPATKAN / impor / $ import_id / warga negara / ulang tahun

1. Dapatkan data untuk penghuni yang memiliki kerabat dari basis data, dan di sisi Python, gabungkan data berdasarkan bulan dan buat daftar untuk bulan-bulan yang tidak ada datanya di dalam basis data.
2. Kompilasi permintaan json di database dan tambahkan bertopik untuk bulan-bulan yang hilang.

DAPATKAN / impor / $ import_id / kota / stat / persentil / usia

1. Dapatkan usia penduduk dari database, dikelompokkan berdasarkan kota, dan kemudian di sisi Python menghitung persentil menggunakan numpy (yang ditentukan sebagai referensi dalam tugas) dan membulatkan hingga dua tempat desimal.
2. PostgreSQL: percentile_cont , SQL-, numpy .

Pengujian

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)

Penangan

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 .

DAPATKAN / impor / $ import_id / warga

• Bongkar dengan beberapa saudara. Memeriksa bahwa untuk setiap penduduk daftar dengan pengidentifikasi kerabat akan dibentuk dengan benar.
• Bongkar dengan satu penduduk tanpa saudara. Memeriksa bahwa bidang tersebut relativesadalah daftar kosong (karena LEFT JOINkueri SQL, daftar kerabat mungkin sama [None]).
• Bongkar dengan seorang residen yang merupakan kerabat dari dirinya sendiri.
• Bongkar kosong. Periksa bahwa pawang memungkinkan untuk menambahkan pembongkaran kosong dan tidak mengalami kesalahan.

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

• Data yang benar, diharapkan berhasil ditambahkan ke database.
  
  
  • ( ).
    
    . , , insert , .
  • ( , ).
    
    , .
    
  • .
    
    , . :)
    
  • 
    
    , aiohttp PostgreSQL 32 767 ( ).
    
  • Empty unloading
    
    Pawang harus memperhitungkan kasing seperti itu dan tidak jatuh, mencoba melakukan pemasukan kosong ke meja dengan penghuni.
    
  
  
• Data dengan kesalahan, mengharapkan respons HTTP 400: Permintaan Buruk.
  
  
  • Tanggal lahir salah (future tense).
  • citizen_id tidak unik dalam unggahan.
  • Hubungan kekerabatan ditunjukkan secara tidak benar (hanya ada dari satu penduduk ke penduduk lainnya, tetapi tidak ada umpan balik).
  • Penduduk memiliki kerabat yang tidak ada dalam bongkar muat.
  • Ikatan keluarga tidak unik.

PATCH / impor / $ import_id / warga negara / $ citizen_id

• Kolom ditentukan, tetapi memiliki tipe data dan / atau format yang salah
• Tanggal lahir salah (waktu mendatang).
• Kolom relativesberisi kerabat yang tidak ada di pembongkaran.

DAPATKAN / impor / $ import_id / warga negara / ulang tahun

• Bongkar di mana seorang penduduk memiliki satu kerabat dalam satu bulan dan dua kerabat di yang lain.
• Bongkar dengan satu penduduk tanpa saudara. Pastikan pawang tidak memperhitungkannya dalam perhitungan.
• Bongkar kosong. Pastikan pawang tidak akan gagal dan akan mengembalikan kamus yang benar dengan 12 bulan sebagai tanggapan.
• Bongkar dengan seorang residen yang merupakan kerabat dari dirinya sendiri. Memeriksa bahwa seorang penduduk akan membeli hadiah untuk bulan kelahirannya.

DAPATKAN / impor / $ import_id / kota / stat / persentil / usia

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

• Bongkar dengan beberapa warga yang ulang tahunnya besok (usia - beberapa tahun dan 364 hari). Periksa apakah prosesor hanya menggunakan jumlah tahun penuh dalam perhitungan.
• Bongkar dengan penduduk yang berulang tahun hari ini (umur - tepatnya beberapa tahun). Ini memeriksa kasus regional - usia penduduk yang ulang tahunnya hari ini tidak boleh dihitung sebagai dikurangi dengan 1 tahun.
• Bongkar kosong. Pawang tidak boleh jatuh di atasnya.

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

Fakta yang menarik: ketika saya menulis tes ini pada tanggal 29 Februari 2020, saya tiba-tiba berhenti mengeluarkan muatan dengan penduduk karena bug di Faker (2020 adalah tahun kabisat, dan tahun-tahun lain yang Faker pilih tidak selalu kabisat tahun di dalamnya juga bukan 29 Februari). Ingatlah untuk mencatat tanggal dan kasus uji tepi!

Migrasi

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

Majelis

1. Menggunakan gambar "berat" snakepacker/python:all(~ 1 GB, ~ 500 MB terkompresi), buat lingkungan virtual, instal semua dependensi dan paket aplikasi ke dalamnya. Gambar ini diperlukan khusus untuk perakitan, dapat berisi kompiler, semua perpustakaan yang diperlukan dan file dengan header.
   
   

   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. Kami menyalin lingkungan virtual yang sudah jadi ke gambar "ringan" snakepacker/python:3.8(~ 100 MB, terkompresi ~ 50 MB), yang hanya berisi penerjemah dari versi Python yang diperlukan.
   
   Penting: di lingkungan virtual, jalur absolut digunakan, sehingga harus disalin ke alamat yang sama saat dirakit di wadah kolektor.
   
   

   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

Menyebarkan

[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

Tentang pengujian stres

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

Apa lagi yang bisa dilakukan?

* * *

Ucapan Terima Kasih