🏛️ NaloGO

Асинхронная Python библиотека для работы с API сервиса самозанятых "Мой налог" (lknpd.nalog.ru)

Порт PHP библиотеки shoman4eg/moy-nalog на асинхронный Python с полной типизацией.

Ключевые возможности

Аутентификация

• ИНН/пароль - классическая аутентификация
• SMS-аутентификация - безопасный вход по номеру телефона
• Автообновление токенов - прозрачная ротация при истечении
• Персистентное хранение - сохранение токенов в файл

Управление доходами

• Создание чеков - одиночные позиции и множественные услуги
• Юридические лица - поддержка корпоративных клиентов
• Отмена чеков - с валидацией причин отмены
• Точная арифметика - decimal.Decimal для финансовых расчетов

Работа с чеками

• JSON данные - полная информация о чеке
• URL печати - прямые ссылки для печати чеков
• Валидация данных - автоматическая проверка корректности

Дополнительные API

• Профиль пользователя - информация об аккаунте
• Способы оплаты - управление банковскими картами
• Налоговая отчетность - история и платежи по ОКТМО

Качество и безопасность

• Покрытие тестами (актуальная цифра в бейдже выше)
• Типизация mypy - статическая проверка типов
• Безопасное логирование - маскировка чувствительных данных
• CI/CD pipeline - автоматические проверки качества

Установка

Из PyPI (рекомендуется)

pip install nalogo

Для разработки

git clone https://github.com/Rusik636/nalogo.git
cd nalogo
pip install -e ".[dev]"

Быстрый старт

Базовая настройка

import asyncio
from nalogo import Client

# Простая инициализация
client = Client()

# С настройками
client = Client(
    base_url="https://lknpd.nalog.ru/api",  # Кастомный endpoint
    storage_path="./tokens.json",           # Файл для токенов
    device_id="my-device-123",              # Кастомный ID устройства
    timezone="Europe/Moscow"                # Зона для дат без смещения
)

Часовой пояс

Налоговая записывает чеки по московскому времени и игнорирует смещение в переданной строке — читает «настенные» цифры как МСК. Поэтому библиотека всегда отправляет время, приведённое к Europe/Moscow. Это не настраивается: любое другое значение на проводе означало бы чек не в тот час.

Параметр timezone управляет только одним: как понимать время без смещения.

Приём оплаты из разных регионов

Рекомендуемый способ — ISO 8601 строка со смещением. Не нужно ни собирать datetime, ни импортировать ZoneInfo, ни настраивать клиент: смещение в строке однозначно задаёт момент, поэтому настройка timezone к нему не применяется и ошибиться нельзя.

income_api = client.income()

# Платёж принят во Владивостоке в 12:00 местного
await income_api.create("Услуга", 5000, operation_time="2025-12-28T12:00:00+10:00")
# -> 2025-12-28T05:00:00+03:00

# Тот же чек, но из Екатеринбурга
await income_api.create("Услуга", 5000, operation_time="2025-12-28T12:00:00+05:00")
# -> 2025-12-28T10:00:00+03:00

# Время лежит в БД в UTC
await income_api.create("Услуга", 5000, operation_time="2025-12-28T12:00:00Z")
# -> 2025-12-28T15:00:00+03:00

Принимаются все формы, которые понимает datetime.fromisoformat():

Объекты datetime работают как прежде:

from datetime import datetime
from zoneinfo import ZoneInfo

op = datetime(2025, 12, 28, 12, 0, tzinfo=ZoneInfo("Asia/Vladivostok"))
await income_api.create("Услуга", 5000, operation_time=op)

Для значений без смещения зону можно задать на конкретный вызов:

await income_api.create(
    "Услуга", 5000,
    operation_time="2025-12-28T12:00:00",
    timezone="Asia/Vladivostok",   # перекрывает настройку клиента
)

Что выбирать:

Ошибки времени

Некорректные значения отсекаются до обращения к сети:

from nalogo import Client, DateTimeFormatException, InputException, TimezoneException

Client(timezone="Moscow")                                  # TimezoneException
await income_api.create(..., operation_time="28.12.2025")  # DateTimeFormatException

Оба класса наследуются от InputException, а тот — от DomainException и ValueError, поэтому ловить можно на любом уровне:

try:
    ...
except InputException:      # только ошибки ввода, до запроса
    ...
except DomainException:     # плюс все ошибки API
    ...

⚠️ До версии 1.1.0 время принудительно переводилось в UTC и отправлялось с суффиксом Z. Налоговая читала эти цифры как московские, из-за чего чек записывался со сдвигом на −3 часа. Если вы обходили это, прибавляя 3 часа вручную, — уберите такую компенсацию после обновления.

Аутентификация

По ИНН и паролю

async def auth_with_inn():
    client = Client()

    # Получение токена
    token = await client.create_new_access_token("123456789012", "your_password")

    # Активация клиента
    await client.authenticate(token)

    print("✅ Аутентификация успешна!")
    return client

По номеру телефона (SMS)

async def auth_with_phone():
    client = Client()

    # Шаг 1: Запрос SMS кода
    phone = "79001234567"
    challenge = await client.create_phone_challenge(phone)

    print(f"📱 SMS код отправлен. Токен: {challenge['challengeToken']}")

    # Шаг 2: Ввод SMS кода (получаете от пользователя)
    sms_code = input("Введите SMS код: ")

    # Шаг 3: Верификация и получение токена
    token = await client.create_new_access_token_by_phone(
        phone, challenge['challengeToken'], sms_code
    )

    # Шаг 4: Активация клиента
    await client.authenticate(token)

    print("✅ SMS аутентификация успешна!")
    return client

Создание чеков

Простой чек

async def create_simple_receipt():
    client = await auth_with_inn()  # Предполагаем аутентификацию

    income_api = client.income()

    result = await income_api.create(
        name="Консультационные услуги",
        amount=5000.00,  # Автоматически конвертируется в Decimal
        quantity=1
    )

    receipt_uuid = result["approvedReceiptUuid"]
    print(f"✅ Чек создан: {receipt_uuid}")

    return receipt_uuid

Чек с несколькими позициями

from nalogo.dto.income import IncomeServiceItem
from decimal import Decimal

async def create_multi_item_receipt():
    client = await auth_with_inn()
    income_api = client.income()

    # Создаем позиции
    services = [
        IncomeServiceItem(
            name="Разработка веб-сайта",
            amount=Decimal("50000.00"),
            quantity=Decimal("1")
        ),
        IncomeServiceItem(
            name="Техподдержка",
            amount=Decimal("5000.00"),
            quantity=Decimal("3")  # 3 месяца
        )
    ]

    result = await income_api.create_multiple_items(services)

    # Проверяем общую сумму: 50000 + (5000 * 3) = 65000
    total = sum(item.amount * item.quantity for item in services)
    print(f"💰 Общая сумма: {total}")

    return result["approvedReceiptUuid"]

Чек для юридического лица

from nalogo.dto.income import IncomeClient, IncomeType

async def create_legal_entity_receipt():
    client = await auth_with_inn()
    income_api = client.income()

    # Информация о юридическом лице
    legal_client = IncomeClient(
        contact_phone="+79001234567",
        display_name="ООО 'Инновационные решения'",
        income_type=IncomeType.FROM_LEGAL_ENTITY,
        inn="1234567890"  # ИНН организации
    )

    result = await income_api.create(
        name="Разработка ПО по договору",
        amount=250000.00,
        quantity=1,
        client=legal_client
    )

    print(f"🏢 Корпоративный чек: {result['approvedReceiptUuid']}")
    return result

Отмена чеков

from nalogo.dto.income import CancelCommentType

async def cancel_receipt():
    client = await auth_with_inn()
    income_api = client.income()

    receipt_uuid = "your-receipt-uuid"

    result = await income_api.cancel(
        receipt_uuid=receipt_uuid,
        comment_type=CancelCommentType.INCORRECT_DATA,
        request_time=datetime.now(timezone.utc)
    )

    print(f"❌ Чек отменен: {result}")

Получение данных чеков

async def get_receipt_info():
    client = await auth_with_inn()
    receipt_api = client.receipt()

    receipt_uuid = "your-receipt-uuid"

    # Получение JSON данных
    receipt_data = await receipt_api.json(receipt_uuid)
    print(f"📋 Сумма: {receipt_data.get('totalAmount')}")
    print(f"📅 Дата: {receipt_data.get('operationTime')}")

    # Генерация URL для печати
    print_url = receipt_api.print_url(receipt_uuid)
    print(f"🖨️ Печать: {print_url}")

Дополнительные API

Информация о пользователе

async def get_user_info():
    client = await auth_with_inn()
    user_api = client.user()

    user_data = await user_api.get()

    print(f"👤 Пользователь: {user_data['displayName']}")
    print(f"📋 ИНН: {user_data['inn']}")
    print(f"📧 Email: {user_data.get('email', 'Не указан')}")
    print(f"📱 Телефон: {user_data['phone']}")

Способы оплаты

async def manage_payment_types():
    client = await auth_with_inn()
    payment_api = client.payment_type()

    # Получение всех способов оплаты
    payment_types = await payment_api.table()
    print(f"💳 Найдено {len(payment_types)} способов оплаты")

    # Поиск избранного способа
    favorite = await payment_api.favorite()
    if favorite:
        print(f"⭐ Избранный: {favorite['bankName']}")
    else:
        print("⭐ Избранный способ не настроен")

Налоговая информация

async def get_tax_info():
    client = await auth_with_inn()
    tax_api = client.tax()

    # Текущие налоги
    tax_data = await tax_api.get()
    print("📊 Налоговая информация получена")

    # История по ОКТМО
    history = await tax_api.history(oktmo="12345678")
    print(f"📈 История операций получена")

    # Платежи (только оплаченные)
    payments = await tax_api.payments(oktmo="12345678", only_paid=True)
    print(f"💸 История платежей получена")

Сетевые сбои и токены

Библиотека различает два исхода, которые нельзя путать: ФНС ответила и отказала — чинится действием человека; ФНС не ответила — чинится повтором.

from nalogo import (
    Client, DomainException, NetworkException,
    RateLimitException, ServiceUnavailableException, UnauthorizedException,
)

try:
    await client.income().create("Услуга", 5000)
except NetworkException:
    # Ответа не было. Доступ цел, просить перепривязать кабинет не надо.
    ...
except RateLimitException as e:
    await asyncio.sleep(e.retry_after or 60)
except ServiceUnavailableException:
    # 502/503/504 — плановые работы или шлюз. Повторить позже.
    ...
except UnauthorizedException:
    # Вот теперь действительно нужна повторная авторизация.
    ...
except DomainException:
    ...

NetworkException наследует DomainException, поэтому общий обработчик ловит и его.

Повторы

Идемпотентные запросы (GET) повторяются до трёх раз с экспоненциальной задержкой и джиттером. Выдача чека — нет: POST /income не идемпотентен на стороне ФНС, слепой повтор мог бы выписать второй документ. Он повторяется только тогда, когда запрос заведомо не ушёл — соединение не установилось:

except ConnectionException:
    # request_may_have_been_sent is False — сервер запроса не видел
    ...
except TimeoutException:
    # Запрос ушёл, ответ не пришёл. Чек мог быть создан — проверьте,
    # прежде чем повторять.
    ...

Отключить повторы: client.http_client.max_attempts = 1.

Токены

Обновление упреждающее — за минуту до истечения, по полю tokenExpireIn. Лишнего запроса с гарантированным 401 не происходит. Параллельные запросы, получившие 401, вызывают ровно одно обновление: ФНС ротирует refresh-токен, и второе обновление ушло бы по уже использованному.

Освобождение соединений

Клиент держит пул соединений открытым для переиспользования:

async with Client(storage_path="./tokens.json") as client:
    ...
# либо явно: await client.aclose()

Безопасность

Хранение токенов

# ❌ Небезопасно - токены в памяти
client = Client()

# ✅ Рекомендуется - сохранение в файл
client = Client(storage_path="./secure_tokens.json")

# ✅ Продакшн - переменные окружения
import os
from pathlib import Path

token_path = Path(os.getenv("TOKEN_STORAGE_PATH", "./tokens.json"))
client = Client(storage_path=str(token_path))

Логирование

import logging

# Настройка логирования для отладки
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("nalogo")

# Библиотека автоматически маскирует чувствительные данные:
# - Токены доступа
# - Пароли
# - Номера телефонов
# - Персональные данные

Обработка ошибок

from nalogo.exceptions import (
    UnauthorizedException,
    ValidationException,
    UnprocessableEntityException,
    DomainException
)

async def safe_operation():
    try:
        client = Client()
        token = await client.create_new_access_token("inn", "password")
        await client.authenticate(token)

    except UnauthorizedException:
        print("❌ Неверный ИНН или пароль")
    except ValidationException as e:
        print(f"❌ Ошибка валидации: {e}")
    except UnprocessableEntityException as e:
        print(f"📱 Ошибка SMS: {e}")
    except DomainException as e:
        print(f"🚨 API ошибка: {e}")
        # e.response содержит httpx.Response для детального анализа

Конфигурация

Переменные окружения

Создайте файл .env:

# API настройки
NALOG_BASE_URL=https://lknpd.nalog.ru/api
NALOG_DEVICE_ID=my-unique-device-id

# Хранение токенов
TOKEN_STORAGE_PATH=./secure/tokens.json

# Аутентификация
NALOG_INN=123456789012
NALOG_PASSWORD=your_secure_password

Использование:

import os
from dotenv import load_dotenv

load_dotenv()

client = Client(
    base_url=os.getenv("NALOG_BASE_URL"),
    device_id=os.getenv("NALOG_DEVICE_ID"),
    storage_path=os.getenv("TOKEN_STORAGE_PATH")
)

Кастомизация HTTP клиента

from nalogo import Client
from nalogo._http import AsyncHTTPClient

# Клиент с кастомными настройками
class CustomHTTPClient(AsyncHTTPClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Увеличиваем таймаут
        self._client.timeout = 60.0

# Использование
client = Client()
client.http_client = CustomHTTPClient("https://lknpd.nalog.ru/api")

Тестирование

Установка зависимостей для разработки

pip install -e ".[dev]"

Запуск тестов

# Все тесты
pytest

# С покрытием
pytest --cov=nalogo --cov-report=html

# Конкретные тесты
pytest tests/test_auth_async.py -v

# Асинхронные тесты
pytest tests/test_income_async.py::TestIncomeAPI::test_create_success -v

Запуск примеров

# Полный пример использования
python examples/async_example.py

# Локальные тесты с моками
python demo.py

Миграция с PHP библиотеки

Соответствие API

Основные различия

1. Асинхронность

// PHP - синхронный код
$result = $client->income()->create($name, $amount, $quantity);

# Python - асинхронный код
result = await client.income().create(name, amount, quantity)

2. Типизация

// PHP - динамическая типизация
$amount = "100.50"; // Строка
$quantity = 2; // Число

# Python - строгая типизация
from decimal import Decimal

amount = Decimal("100.50")  # Decimal для точности
quantity = Decimal("2")     # Decimal для консистентности

3. Обработка ошибок

// PHP - исключения базового класса
try {
    $result = $client->income()->create(...);
} catch (DomainException $e) {
    // Общая обработка
}

# Python - специфичные исключения
try:
    result = await client.income().create(...)
except ValidationException as e:
    # Конкретная ошибка валидации
except UnauthorizedException as e:
    # Ошибка авторизации

Шаблон миграции

# Шаблон для миграции PHP кода
async def migrate_from_php():
    # 1. Замените синхронный клиент на асинхронный
    # PHP: $client = new ApiClient();
    client = Client()

    # 2. Добавьте await ко всем API вызовам
    # PHP: $token = $client->createNewAccessToken($inn, $password);
    token = await client.create_new_access_token(inn, password)

    # 3. Замените ассоциативные массивы на объекты DTO
    # PHP: $client = ['contactPhone' => $phone, ...];
    from nalogo.dto.income import IncomeClient
    client_data = IncomeClient(contact_phone=phone, ...)

    # 4. Используйте Decimal для денежных операций
    # PHP: $amount = 100.50;
    from decimal import Decimal
    amount = Decimal("100.50")

    # 5. Обновите обработку исключений
    # PHP: catch (DomainException $e)
    # Python: except DomainException as e

Производительность

Бенчмарки

Оптимизация для высоких нагрузок

import asyncio
from nalogo import Client

async def bulk_receipts():
    client = await auth_with_inn()
    income_api = client.income()

    # Создание множества чеков параллельно
    tasks = []
    for i in range(100):
        task = income_api.create(f"Услуга {i}", 1000.00, 1)
        tasks.append(task)

    # Выполнение всех задач параллельно
    results = await asyncio.gather(*tasks, return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception))
    print(f"✅ Создано {success_count} из {len(tasks)} чеков")

Известные ограничения

API ограничения

• Invoice API не реализован (помечен как "Not implemented" в оригинальной PHP библиотеке)
• API версии v1/v2 endpoints могут иметь различия в поведении
• Лимиты запросов определяются сервисом Мой Налог

Совместимость

• Python 3.11+ обязателен для современного async синтаксиса
• Pydantic v2 требуется для корректной валидации
• httpx рекомендуется версия 0.25.0+

Вклад в развитие

Настройка окружения разработки

# Клонирование
git clone https://github.com/Rusik636/nalogo.git
cd nalogo

# Создание виртуального окружения
python -m venv .venv
source .venv/bin/activate  # Linux/Mac
# или
.venv\Scripts\activate     # Windows

# Установка в режиме разработки с версиями инструментов как в CI
pip install -r requirements-dev.txt
pip install -e . --no-deps

# Настройка pre-commit хуков (перезапишет старый самописный хук, если он есть)
pre-commit install

Запуск проверок качества

# Линтинг
ruff check .

# Форматирование
black .

# Типизация
mypy nalogo/

# Безопасность
bandit -r nalogo/

# Полная проверка (как в CI)
pytest --cov=nalogo --cov-fail-under=80

Создание PR

1. Создайте ветку для фичи: git checkout -b feature/amazing-feature
2. Напишите тесты для новой функциональности
3. Убедитесь что все проверки проходят
4. Создайте PR с подробным описанием изменений

Лицензия

MIT License - подробности в файле LICENSE.

Благодарности

Оригинальная PHP библиотека — Artem Dubinin (shoman4eg/moy-nalog).

Поддержка

• 📋 Issues: GitHub Issues
• 📖 Документация: README.md
• 💬 Обсуждения: GitHub Discussions
• 📧 Email: ruslan.prokshin@mail.ru