OpenAPI tools

Библиотека позволяет без изменений в логике публиковать её методы в спецификации и сервере

Basic usage

Пример 0: эндпоинт без аргументов

Допустим, мы хотим опубликовать метод по извлечению корня из целых чисел

import time
import asyncio

from http_tools import HttpServer
from openapi_tools import OpenApiServer, RpcEndpoint

# Обязательно указание типа каждого аргумента и результата
def get_now() -> float:
    return time.time()

async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='readme'))
    api_server.register_endpoint(RpcEndpoint(get_now, "POST", "/time/now", [], {}))
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Пример 1: базовое использование

Допустим, мы хотим опубликовать метод по извлечению корня из целых чисел

import math
import asyncio

from http_tools import HttpServer
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    return math.sqrt(value)

async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='readme'))
    api_server.register_endpoint(gen_json_rpc_endpoint(sqrt, "POST", "/math/sqrt", [], {}))
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Можно проверить эндпоинт с помощью запроса из консоли:

curl -X POST --location "http://127.0.0.1:8888/math/sqrt" -H "Content-Type: application/json" -d '{"value": 42}'

Получим следующий ответ:

{"done": true, "result": 6.48074069840786}

Если пошлём некорректный тип:

curl -X POST --location "http://127.0.0.1:8888/math/sqrt" -H "Content-Type: application/json" -d '{"value": "bad"}'

то библиотека попробует скастить его в требуемый тип и мы получим следующий ответ:

{"error": "{'value': ValueError(\"invalid literal for int() with base 10: 'bad'\")}", "error_type": "FieldErrors", "error_code": null, "done": false}

Однако, если мы сделаем запрос с отрицательным числом

curl -X POST --location "http://127.0.0.1:8888/math/sqrt" -H "Content-Type: application/json" -d '{"value": -1}'

то получим код 500 и довольно невнятную ошибку:

{"error": "math domain error", "error_type": "ValueError", "error_code": null, "done": false}

Пример 2: Обработка исключений

Доработаем обработку ошибок, для этого

• добавим проверку чисел на отрицательность
• при регистрации обработчика добавим ответ для возникающего типа исключения

import math
import asyncio

from http_tools import HttpServer, HttpStatusCode
from http_tools.answer import ExceptionAnswer
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    assert value >= 0, f'Only non negative values allowed, got: {value}'
    return math.sqrt(value)

async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme"))
    api_server = OpenApiServer(OpenApiServer.Config(), OpenApiServer.Context(
        http_server=http_server, company='ABM', 
        project_group='openapi_tools', project_name='readme'))
    api_server.register_endpoint(gen_json_rpc_endpoint(sqrt, "post", "/math/sqrt", [], {
        AssertionError: ExceptionAnswer[HttpStatusCode.BadRequest]
    }))
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Проверим ответ на отрицательное число:

curl -X POST --location "http://127.0.0.1:8888/math/sqrt" -H "Content-Type: application/json" -d '{"value": -1}'

Получим код 400 и более понятную ошибку:

{"error": "Only non negative values allowed, got: -1", "error_type": "AssertionError", "error_code": null, "done": false}

Если же требуется автоматизировать обработку различных ошибок на клиенте, то в ответе стоит использовать ErrorCode

Пример 3: Использование кодов ошибок

Для демонстрации допустим, что sqrt не может вычислять корень слишком больших значений Создадим отдельные исключения и для каждого из них укажем ErrorCode

import math
import asyncio

from http_tools import HttpServer, HttpStatusCode
from http_tools.answer import ExceptionAnswer, ErrorCode
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint

class TooBigValue(ValueError):
    pass

class NegativeValue(ValueError):
    pass

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    if value > 100:
        raise TooBigValue(f'Only values lower than 100 allowed, got: {value}')
    if value < 0:
        raise NegativeValue(f'Only non negative values allowed, got: {value}')
    return math.sqrt(value)

async def main():
        http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme"))
        api_server = OpenApiServer(
            OpenApiServer.Config(), OpenApiServer.Context(
                http_server=http_server, company='ABM', project_group='openapi_tools', project_name='readme'))
        api_server.register_endpoint(gen_json_rpc_endpoint(sqrt, "POST", "/math/sqrt", [], {
            TooBigValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(1)],
            NegativeValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(2)],
        }))
        await http_server.async_init()
        await asyncio.sleep(10 ** 10)

asyncio.run(main())

Проверим ответ на отрицательное число:

curl -X POST --location "http://127.0.0.1:8888/math/sqrt" -H "Content-Type: application/json" -d '{"value": -1}'

Получим код 400 и ошибку с кодом:

{"error": "Only non negative values allowed, got: -1", "error_type": "NegativeValue", "error_code": 2, "done": false}

Проверим ответ на большое число:

curl -X POST --location "http://127.0.0.1:8888/math/sqrt" -H "Content-Type: application/json" -d '{"value": 111}'

Получим код 400 и ошибку с кодом:

{"error": "Only values lower than 100 allowed, got: 111", "error_type": "TooBigValue", "error_code": 1, "done": false}

Однако, данный подход(коды ошибок указываются напрямую) может приводить к путанице при увеличении размеров API - в разных эндпоинтах один и тот же код может обозначать разное.

Пример 4: Наделение одного кода ошибки разными смыслами

Проиллюстрируем проблему, добавив ещё одну функцию бизнес-логики по возведению числа в степень, и опубликуем её в апи

import math
import asyncio

from http_tools import HttpServer, HttpStatusCode
from http_tools.answer import ExceptionAnswer, ErrorCode
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint

class TooBigValue(ValueError):
    pass

class NegativeValue(ValueError):
    pass

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    if value > 100:
        raise TooBigValue(f'Only values lower than 100 allowed, got: {value}')
    if value < 0:
        raise NegativeValue(f'Only non negative values allowed, got: {value}')
    return math.sqrt(value)


# Обязательно указание типа каждого аргумента и результата
def power(base: float, exponent: float) -> float:
    if base < 0 and int(exponent) != exponent:
        raise NegativeValue(f'With fractional exponent({exponent}) only non negative bases allowed, got: {base}')
    return math.pow(base, exponent)


async def main():
        http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme"))
        api_server = OpenApiServer(OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', 
            project_group='openapi_tools', project_name='readme'))
        api_server.register_endpoint(gen_json_rpc_endpoint(sqrt, "post", "/math/sqrt", [], {
            TooBigValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(1)],
            NegativeValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(2)],
        }))
        api_server.register_endpoint(gen_json_rpc_endpoint(power, "post", "/math/pow", [], {
            NegativeValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(1)],
        }))
        await http_server.async_init()
        await asyncio.sleep(10 ** 10)

asyncio.run(main())

Проверим эндпоинт /math/pow:

curl -X POST --location "http://127.0.0.1:8888/math/pow" -H "Content-Type: application/json" -d '{"base": -1, "exponent": -2.1}'

Получим код 400 и ошибку с кодом 1:

{"error": "With fractional exponent(-2.1) only non negative bases allowed, got: -1.0", "error_type": "NegativeValue", "error_code": 1, "done": false}

При этом получается, что одна и та же проблема в разных эндпоинтах имеет разные коды, что может путать разработчиков и усложнять код клиентского приложения.
Есть два подхода для составления API, более единообразного в плане кодов ошибок:

1. Именованный реестр кодов ошибок
2. Единый словарь обработки исключений

Пример 5: Именованный реестр кодов

Создадим enum со своим перечнем кодов ошибок, тогда при использовании это будут не просто числа, а именованные параметры

import enum
import math
import asyncio

import http_tools
from http_tools import HttpServer, HttpStatusCode
from http_tools.answer import ExceptionAnswer
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint

class TooBigValue(ValueError):
    pass

class NegativeValue(ValueError):
    pass

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    if value > 100:
        raise TooBigValue(f'Only values lower than 100 allowed, got: {value}')
    if value < 0:
        raise NegativeValue(f'Only non negative values allowed, got: {value}')
    return math.sqrt(value)

@enum.unique
class ErrorCode(http_tools.answer.ErrorCode, enum.Enum):
    negative_value = 1
    too_big_value = 2

# Обязательно указание типа каждого аргумента и результата
def power(base: float, exponent: float) -> float:
    if base < 0 and int(exponent) != exponent:
        raise NegativeValue(f'With fractional exponent({exponent}) only non negative bases allowed, got: {base}')
    return math.pow(base, exponent)

async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme:example1"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='example1'))
    api_server.register_endpoint("POST", "/math/sqrt", gen_json_rpc_endpoint(sqrt, [], {
        TooBigValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode.too_big_value],
        NegativeValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode.negative_value],
    }))
    api_server.register_endpoint("POST", "/math/pow", gen_json_rpc_endpoint(power, [], {
        NegativeValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode.negative_value],
    }))
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Такой подход требует минимальных изменений и увеличивает читаемость, но и с ним возможны ошибки

Example 6

Единый словарь обработки исключений Мы можем собрать единый словарь со всеми типами исключений и выбирать из него необходимые для каждого эндпоинта
ВНИМАНИЕ, передавать этот словарь целиком при каждой регистрации обработчика НЕЛЬЗЯ Это приведёт к формированию спецификации с огромным количеством лишних ответов

import math
import typing
import asyncio

from http_tools import HttpServer, HttpStatusCode
from http_tools.answer import ExceptionAnswer, ErrorCode
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint

class TooBigValue(ValueError):
    pass

class NegativeValue(ValueError):
    pass

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    if value > 100:
        raise TooBigValue(f'Only values lower than 100 allowed, got: {value}')
    if value < 0:
        raise NegativeValue(f'Only non negative values allowed, got: {value}')
    return math.sqrt(value)


# Обязательно указание типа каждого аргумента и результата
def power(base: float, exponent: float) -> float:
    if base < 0 and int(exponent) != exponent:
        raise NegativeValue(f'With fractional exponent({exponent}) only non negative bases allowed, got: {base}')
    return math.pow(base, exponent)

class PickableDIct(dict):
    def pick(self, keys: typing.Iterable) -> dict:
        """Возвращает новый словарь только с выбранными ключами."""
        return {k: self[k] for k in keys if k in self}

async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme:example1"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='example1'))
    exception_to_answer = PickableDIct({
        TooBigValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(1)],
        NegativeValue: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(2)],
    })
    api_server.register_endpoint("POST", "/math/sqrt", gen_json_rpc_endpoint(
        sqrt, [], exception_to_answer.pick({NegativeValue, TooBigValue}))
    )
    api_server.register_endpoint( "POST", "/math/pow", gen_json_rpc_endpoint(
        power, [], exception_to_answer.pick({NegativeValue}))
    )
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Этот способ на данный момент является рекомендуемым при использовании кодов ошибок в ответах

Авторизация

Во всех прошлых примерах можно заметить, что при генерации эндпоинта передаётся пустой массив.
Это список схем авторизации. С его помощью выполняется управление доступами.
Чтобы ограничить доступ к какому-то эндпоинту достаточно передать в него экземпляр любого класса, дочернего к SecurityScheme.
Рассмотрим их по очереди
Код, связанный с исключениями, удалён для уменьшения размера примера.
В боевом коде стоит использовать и security, и исключения

import math
import asyncio

from http_tools import HttpServer
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint, ParameterLocation, Security

# Обязательно указание типа каждого аргумента и результата
def sqrt(value: int) -> float:
    return math.sqrt(value)
    
async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme:example1"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='example1'))
    # Создадим схему безопасности, требующую передачу токена в query части запроса под именем 'name'
    # Для простоты примера валидным значением токена будем считать только "secret", в бою так делать не стоит  
    api_key_scheme = ApiKeySecurityScheme(in_=ParameterLocation.query, name='token', resolver=lambda x, s: x == 'secret')
    api_server.register_endpoint("POST", "/math/sqrt", gen_json_rpc_endpoint(
        sqrt, [Security(api_key_scheme)], {})
    )
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Но есть кейсы, когда методу бизнес-логики требуется знать перечень объектов, который является правовой информацией пользователя, например: получение списка камер в БР фильтрует их по регионам, доступным пользователю Т

import math
import typing
import asyncio
from dataclasses import dataclass

from http_tools import HttpServer
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint, ParameterLocation, Security

@dataclass
class Camera:
    id: int
    region_id: int

all_cameras = [Camera(i, i % 3) for i in range(11)]
# Обязательно указание типа каждого аргумента и результата
def list_camera(region_restrictions: list[int] | None) -> list[Camera]:
    if region_restrictions is None:
        return all_cameras
    return list(filter(lambda c: c.region_id in region_restrictions, all_cameras)) 

class PickableDIct(dict):
    def pick(self, keys: typing.Iterable) -> dict:
        """Возвращает новый словарь только с выбранными ключами."""
        return {k: self[k] for k in keys if k in self}
    
async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme:example1"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='example1'))
    
    def resolver(token: str, scopes: list[str]) -> dict:
        return PickableDIct({
            'first_user_token': {"region:": [1]}, 
            'second_user_token': {"region:": [2]}, 
            'admin_token': {"region:": [1,2,3]}, 
            'root_token': {"region:": None}
        }.get(token)).pick(scopes)
    # Создадим схему безопасности, требующую передачу токена в query части запроса под именем 'name'
    api_key_scheme = ApiKeySecurityScheme(in_=ParameterLocation.query, name='token', resolver=resolver)
    api_server.register_endpoint("POST", "/camera/list", gen_json_rpc_endpoint(
        list_camera, [Security(api_key_scheme, {'region_restrictions': "region:"})], {})
    )
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())

Финальный пример

Рассмотрим большой и сложный пример - полный процесс согласования заявки(Proposal):

• Заявки создаются, редактируются и отправляются на подписание исполнителями
• Заявки просматриваются и подписываются, либо возвращаются на доработку подписантами
• Подписанные заявки применяются, либо возвращаются на доработку админами
• Кроме того, для техподдержки требуются расширенные возможности

Выделим права каждой из групп
Первая группа:

• action:proposal:read просмотр своих заявок
• action:proposal:write создание, изменение и удаление своих заявок
• action:proposal:send_to_signing отправка на подписание

Вторая группа:

• action:proposal:read_all просмотр всех заявок
• action:proposal:sign подписание
• action:proposal:rollback возврат на доработку

Третья группа:

• action:proposal:read_all просмотр всех заявок
• action:proposal:rollback возврат на доработку
• action:proposal:apply применение заявки

Четвёртая группа:

• action:proposal:read_all просмотр всех заявок
• action:proposal:write_all создание, изменение и удаление всех заявок
• action:proposal:write_signing изменение и удаление заявки на этапе подписания

import enum
import typing
import asyncio
from typing import Literal
from dataclasses import dataclass

from http_tools import HttpServer
from openapi_tools import OpenApiServer, gen_json_rpc_endpoint, ParameterLocation, Security, ApiKeySecurityScheme

@dataclass(kw_only=True)
class Proposal:
    id: int | None = None
    content: dict
    creator_id: int
    signing_at: float | None = None
    signed_by_id: int | None = None
    applied_at: float | None = None
    
class ProposalStatus(enum.Enum):
    draft = Proposal.signing_at==None
    signing = and_(Proposal.signing_at!=None, Proposal.signed_by_id==None)
    signed = and_(Proposal.signed_by_id!=None, Proposal.applied_at==None)
    applied = and_(Proposal.applied_at!=None)

class Controller:
    def add_proposal(self, content: dict, creator_id: int) -> int:
        return self.context.db.add(Proposal(content=content, creator_id=creator_id))
    
    def list_proposal(
            self, filter_by_creator_id: int | None, limit: int, offset: int,
            filter_by_status: ProposalStatus | None = None  
    ) -> list[Proposal]:
        query = self.context.db.get(Proposal)
        query = query if filter_by_creator_id is None else query.filter(Proposal.creator_id==filter_by_creator_id)
        query = query if filter_by_status is None else query.filter(filter_by_status.value)
        return query.order_by(Proposal.id).limit(limit).offset(offset)
    
    def get_proposal(self, id: int, filter_by_creator_id: int | None) -> Proposal:
        query = self.context.db.get(Proposal).filter(Proposal.id==id)
        query = query if filter_by_creator_id is None else query.filter(Proposal.creator_id==filter_by_creator_id)
        return query.one()
    
    def update_proposal(self, id: int, content: dict, filter_by_creator_id: int | None, update_signing: bool) -> None:
        # Проверим наличие доступа до заявки, если нет, то вылетит исключение
        self.get_proposal(id=id, filter_by_creator_id=filter_by_creator_id)
        query = self.context.db.update(Proposal).filter(Proposal.id==id).set(content=content)
        # Не всем разрешено обновлять не свои заявки
        query = query if filter_by_creator_id is None else query.filter(Proposal.creator_id==filter_by_creator_id)
        # Не всем разрешено обновлять заявку во время согласования
        query = query if update_signing else query.filter(Proposal.signed_by_id==None) 
        query = query.filter(Proposal.applied_at==None)  # Всем запрещено обновлять принятие заявки
        query.execute()

class PickableDIct(dict):
    def pick(self, keys: typing.Iterable) -> dict:
        return {k: self[k] for k in keys if k in self}
    
async def main():
    http_server = HttpServer(HttpServer.Config(port=8888), HttpServer.Context(instance_id="readme:example1"))
    api_server = OpenApiServer(
        OpenApiServer.Config(), OpenApiServer.Context(
            http_server=http_server, company='ABM', project_group='openapi_tools', project_name='example1'))
    controller = Controller()
    
    class AuthInfo(...):
        scopes: set[str]
        user_id: int
    
    def resolver(token: str) -> tuple[Jsonable, set[str]]:
        auth_info = {
            'first_user_token': {'user_id': 1, 'action': ['proposal:write', 'proposal:read']}, 
            'second_user_token': {'user_id': 2, 'action': ['proposal:write', 'proposal:read']}, 
            'approver_token': {'user_id': 3, 'action': ['proposal:write', 'proposal:read', 'proposal:read_all', 'proposal:write_all']}, 
            'admin_token': {},
        }.get(token)
        return auth_info, auth_info['action']
        # return PickableDIct({
        #     'first_user_token': {'user_id': 1, 'action': {'proposal': ['write', 'read']}}, 
        #     'second_user_token': {'user_id': 2, 'action': {'proposal': ['write', 'read']}}, 
        #     'approver_token': {'user_id': 3, 'action': {'proposal': ['write', 'read', 'read_all', 'write_all']}}, 
        #     'admin_token': {},
        # }.get(token)).pick(scopes)
    
    # outer_auth_resolver: Callable[[token], info]
    # resolver=immovable.resolve_token
    # Создадим схему безопасности, требующую передачу токена в query части запроса под именем 'token'
    api_key = ApiKeySecurityScheme[AuthInfo](
        location=ParameterLocation.query, name='token', resolver=resolver
    )
    exc_to_answer = PickableDIct({
        NotFound: ExceptionAnswer[HttpStatusCode.BadRequest][ErrorCode(1)],
    })
    # Создадим схему безопасности, требующую передачу токена в header запроса под именем 'server_name'
    server_header = ApiKeySecurityScheme[None](
        location=ParameterLocation.header, name='server_name', resolver=lambda x: None)
    
    api_server.register_endpoint(gen_json_rpc_endpoint(
        controller.add_proposal, "POST", "/proposal/add",  
        securities=[
            Security(server_header).use(filter_by_creator_id=Literal[None]),
            Security(api_key, scopes=['proposal:read_all']).use(filter_by_creator_id=Literal[None]),
            Security(api_key, scopes={'proposal:read'}).use(filter_by_creator_id='_.user_id | $cast.int'),
            Security(api_key, scopes={'proposal:read'}).use(filter_by_creator_id='user_id | to_number(@)'),
            api_key.has('action:proposal:read_all').use(filter_by_creator_id=JPath('user_id')),
            api_key.scopes('action:proposal:read_all').kwargs(filter_by_creator_id=None),
            api_key.scopes('action:proposal:read').kwargs(filter_by_creator_id=lambda info: info['user_id']),
            # api_key, ['action:proposal:read_all', {'filter_by_creator_id': None}]
            
            Security(keycloak, scopes=['camera:write']).use(allowed_camera_ids=None),
        ], 
        exception_type_to_answer_type={}
    ))
    # Разрешаем добавление предложений пользователям с правом action:proposal:write, 
    # при этом передаём значение user_id пользователя в качестве аргумента creator_id
    api_server.register_endpoint("POST", "/proposal/add", gen_json_rpc_endpoint(
        controller.add_proposal, [Security(api_key, ['action:proposal:write'], {'creator_id': 'user_id'})], {}))
    # Разрешаем просмотр предложений пользователям в зависимости от прав:
    # с action:proposal:read_all передаём None в качестве аргумента filter_by_creator_id, чтобы отключить фильтрацию
    # с action:proposal:read передаём user_id пользователя в качестве аргумента filter_by_creator_id, чтобы отобразить только его предложения
    # Порядок играет важную роль - работает только первая успешная Security, т.е. при наличии обоих прав не будет фильтрации
    api_server.register_endpoint("GET", "/proposal", gen_query_rpc_endpoint(
        controller.list_proposal, [
            Security(api_key, ['action:proposal:read_all', {'filter_by_creator_id': None}]),
            Security(api_key, ['action:proposal:read'], {'filter_by_creator_id': 'user_id'})
        ], {}))
    api_server.register_endpoint("GET", "/proposal/{id}", gen_query_rpc_endpoint(
        controller.get_proposal, [Security(api_key, ['action:proposal:write'])], exc_to_answer.pick({NotFound})))
    
    # Разрешаем обновление заявок. Для этого пользователь должен иметь либо 
    # - право write_all и тогда он может обновлять и свои и чужие заявки 
    # - право write и тогда он может обновлять только свои заявки
    # Кроме того, обновление заявок, отправленных на подписание требует отдельного права
    api_server.register_endpoint("PUT", "/proposal/{id}", gen_json_rpc_endpoint(
        controller.update_proposal, [
            Security(api_key, ['action:proposal:write_all', {
                'filter_by_creator_id': None, 'update_signing': 'action:proposal:update_signing'}]),
            Security(api_key, ['action:proposal:write'], {
                'filter_by_creator_id': 'user_id', 'update_signing': 'action:proposal:update_signing'})
        ], {}))
    await http_server.async_init()
    await asyncio.sleep(10 ** 10)

asyncio.run(main())