Общие принципы

API принимает GET и POST HTTP запросы. В случае успешного запроса сервер отвечает статусом 200.

Формат JSON ответа от сервера:

{
	"data": {},
	"errors": [],
	"status": "ok"
}
  • data - [] или {}
  • errors - список ошибок
  • status - ok или bad

При авторизации система выдает sid, все последующие запросы выполняются передачей этого идентификатора. Cрок жизни идентификатора 15 минут. Обновлять его требуется только при получении ошибки WrongSIDException в поле errors.

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

  • Переавторизация перед каждым запросом;
  • Частые и однотипные запросы;

Ошибка DuplicateRequestException в поле errors говорит о том, что клиент повторяет один и тот же запрос слишком часто.

Для обеспечения равномерной нагрузки на сетевое и серверное оборудование, компания SAURES оставляет за собой право ограничивать количество запросов в единицу времени, а также может создавать БАН листы IP адресов нарушителей. Необходимо использовать принцип необходимого и достаточного объема обмена данными с сервером!

Общая схема работы с API

Общая схема работы с API

Пример клиента на Python

Асинхронный клиент на aiohttp/python3.7+ с автоматическим обновлением sid'а анализом ошибок и 5ю повторами.

import asyncio
from aiohttp import ClientSession, ClientOSError, ContentTypeError


class APIClient:
    API_URL = 'https://api.saures.ru/1.0'
    request_attempts = 5  # кол-во попыток
    sid = None

    def __init__(self, email: str, password: str):
        self._email = email
        self._password = password
        self._sid_renewal = False

    async def request(self, method: str, url: str, **kwargs):
        data = {}
        data.update(kwargs)
        for i in range(1, self.request_attempts + 1):
            if self.sid:
                data.update({'sid': self.sid})
            try:
                async with ClientSession() as session:
                    async with session.request(
                        method, self.API_URL + url,
                        params=data if method == 'GET' else None,
                        data=data if method == 'POST' else None,
                    ) as request:
                        response = await request.json()
            except (ClientOSError, ContentTypeError):
                if i == self.request_attempts:
                    return False
                await asyncio.sleep(60)
            else:
                if response.get('status') == 'ok':
                    return response
                errors = [e['name'] for e in response['errors']]
                if 'WrongSIDException' in errors:
                    if not self._sid_renewal:
                        self.sid = None
                    if i == self.request_attempts:
                        return False
                    await self.check_sid()
                    continue
                if 'DuplicateRequestException' in errors:
                    if i == self.request_attempts:
                        return False
                    await asyncio.sleep(10)
                    continue
                return response
        return False

    async def update_sid(self):
        if self._sid_renewal:
            return False
        self._sid_renewal = True
        try:
            response = await self.request(
                'POST', '/login',
                email=self._email, 
                password=self._password
            )
            self.sid = response['data']['sid']
        except:
            self._sid_renewal = False
            return False
        else:
            self._sid_renewal = False
            return True

    async def check_sid(self):
        if self._sid_renewal:
            while self._sid_renewal:
                await asyncio.sleep(1)
        elif not self.sid:
            await self.update_sid()

        if self.sid:
            return True
        return False

    async def user_objects(self):
        if await self.check_sid():
            response = await self.request('GET', '/user/objects')
            if response:
                return response['data']


async def main():
    client = APIClient(email='demo@saures.ru', password='demo')
    print(await client.user_objects())


asyncio.run(main())
© 2023 SAURES - система автоматизированного учета и контроля ресурсов