Django *

Находки на рыбалке: FastStream! Сходили с автором половить рыбу и пожарить всякое вкусное на углях. И пока было время – говорили про программирование.

Для тех, кто еще не успел познакомиться с новинкой, рассказываю: FastStream – аналог FastAPI, но для работы с событиями в брокерах / очередях.

Пример:

from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@publisher("another-topic")
@subscriber("first-topic")
async def handle(user: str) -> str:
     return f"Hi, {user}!"

Что делает данный код? Читает сообщения из first-topic, парсит из них поле user типа str, выполняет вашу логику обработки, отправляет новое сообщение в another-topic. Просто? Удобно? Да!

Что нам дает такой подход?

• Декларативное описание, чего мы хотим. Не надо руками создавать коннекты и рулить потоком выполнения

• AsyncAPI документацию (аналог OpenAPI в вебе)

• Удобное тестирование

• Кучу других плюшек!

Внутри видео обсудили:

• Детали работы DI фастстрима

• Встроенное Observability

• Open Tracing

• Сообщество фреймворка (тут не будет проблемы "одного автора", сообщество живет!)

• Отличия от Celery: почему интструменты совсем разные?

Репозиторий: https://github.com/ag2ai/faststream
Документация: https://faststream.ag2.ai

Обсуждение: используется ли в ваших проектах асинхронная архитектура? Как вы работаете с событиями?

Добавляем подписку с Telegram Stars в свой апп

Я очень радовался 14.08.2024 из‑за выхода очередного релиза Телеграм, в котором они анонсировали подписки за звезды, потому что это сильно облегчало мое взаимодействие с клиентами: они один раз подписываются и далее не имеют проблем с повторными платежами, вместо этого звезды тихонько списываются у них с счетов. В этом посте я расскажу, как прикрутил подписку в звездах к своей Telegram Mini App (TMA).

Вот как выглядит глазами клиентов списание звезд в пользу вашей аппы. Заметьте, никаких предупреждающих писем, как у Stripe, например.

Свои подписки можно посмотреть у себя в профиле. Это выглядит так:

https://storage.googleapis.com/memes-repo/tutorial/100000998111.png

Отменить подписку можно, если вы зайдете в ее детали.

https://storage.googleapis.com/memes-repo/tutorial/1000009980.png

Теперь, когда вы представляете все визуально, то опишу последовательность вызовов, которые вы должны выполнить.

1. Создать счет на подписку.

2. Подтвердить оплату.

3. Принять первое и последующие сообщения об оплате.

### Создание счета

Для создания счета вызываем метод createInvoiceLink.

    def create_invoice(admin_chat_id: str, 
                       channel_id: str, 
                       title: str, 
                       description: str,
                       tariff_id: int, 
                       tariff_price: int,
                       ) -> dict:

        data = {
            "chat_id": admin_chat_id,
            "title": title,
            "description": description,
            "payload": json.dumps({"t": tariff_id}), 
            "provider_token": "",  # Empty
            "currency": "XTR", # required
            "prices": [{'amount': tariff_price, 'label': title}],
            "subscription_period": 2592000,
        }

        r = requests.post(f"{TG_API_URL}{TG_OP_SEND_INVOICE}", json=data)

Вещи, на которые обращаем внимание:

1. provider_token пустая строка

2. currency обязательно строка XTR

3. subscription_period обязательно 2 592 000

4. prices содержит только один элемент

### Подтверждение оплаты

Для получения подтверждения оплаты счета вам нужно добавить в ваш webhook (allowed_updates) подписку на pre_checkout_query. После этого вы начнете получать update c данным полем и структурой.

Когда вы получаете такой апдейт, вам нужно отреагировать на него в течение 10 секунд: принимаете или отклоняете (например, если этот счет уже был оплачен и другие краевые условия).

Данная структура также позволяет вам понять, что было оплачено в рамках счета, путем чтения поля payload (которое было записано при создании счета).

        pre_checkout_data: PreCheckoutData = {
            "id": packet["pre_checkout_query"]["id"],
            "currency": packet["pre_checkout_query"]["currency"],
            "total_amount": packet["pre_checkout_query"]["total_amount"],
            "invoice_payload": json.loads(packet["pre_checkout_query"]["invoice_payload"])
        }

Для подтверждения или отклонения оплаты счета используйте метод answerPreCheckoutQuery.

### Принять сообщения об оплате

На мое удивление сообщение об оплате встроено в существующий объект Message, так что не нужно дополнительно подписываться на новые источники данных. Вместо этого, мы просматриваем все message в поисках поля successful_payment или refunded_payment и записываем в статистику (для возможного возврата в том числе).

Обратите внимание! В структуре SuccessfulPayment есть параметр is_first_recurring, и я думал, что он True для первого платежа подписки и False для последующих, но! в последующих его просто нет (это допустимо, так как это необязательный параметр).

На этом все, как только вы начали получать successful_payment, вам не нужно больше делать ничего, чтобы клиент продлил подписку.

P.S. Вы всегда можете обновлять срок подписки при получении данного сообщения, читая поле subscription_expiration_date. Если successful_payment перестал приходить, то и подписка не обновится.

Ваш Дима из TG Defender, защита вашего Telegram канала от наплыва ботов на стиле.

Как создавать веб-приложение на Django с нуля?

Внутри — четыре инструкции по веб-разработке на Django. Начинаем с создания и настройки простых проектов: блога и канбан-доски. Затем переходим к подключению автоматических бэкапов. Изучение займет не более часа.

После прохождения вы научитесь:

• работать с бэкендом и API,

• создавать веб-приложение по шаблону от Django,

• настраивать Nginx для обработки запросов,  

• развертывать Django-приложение с помощью Gunicorn.

Переходите в Академию Selectel и изучайте курс.

Привет, я Иван Елфимов, Developer Advocate в Островке. Расскажу про самые полезные open source решения, которые использую я сам и Команда партнерских интеграций.

• Вообще, трудно назвать не open source решение, которое бы использовал Островок в своей разработке. Используем большей частью опенсорс с подходящими лицензиями.

• Один из самых малоизвестных и узкопрофильных компонентов, который мы используем: django-chronograph. Это приложение, позволяющее запускать management-команды из админки. Эта библиотека давно не поддерживается, поэтому мы форкнули её к себе доработали. Возможно когда-нибудь выложим в Open Source.

• Компания Astral создает (точнее пересоздает) отдельные инструменты для питона на языке Rust, заставляя их работать в разы быстрее. Мы уже начали пользоваться и ruff и uv.

• Личная рекомендация от меня: pre-commit. Инструмент очень мощный и полезный, хотя даже там встречается юмор от разработчиков. К примеру, скрипт проверки кода на случайно оставленный print называется «oops, I print it again!»

• Островок готовит к выпуску свой техрадар. Пока он работает в закрытом режиме только для сотрудников, но в планах вынести его наружу. В техрадаре будет информация о библиотеках, прошедших через тесты команды Островка. Они будут сгруппированы по направлениям (frontend, backend, ios/android, аналитика, QA, devops) и каждая из библиотек будет иметь свою стадию: adopt, trial, assess и hold.

Привет, я Иван Елфимов, Developer Advocate в Ostrovok.ru. Расскажу про наш переход на Django 5.0.

4 марта вышел третий security release, когда мы обычно переходим на новые версии библиотек. Мы были готовы к переходу: проверили на декабрьском 5.0.0. что могло сломаться. Из-за некоторых обновлений нам пришлось менять код:

• Раньше мы использовали nulls_last=False. Теперь nulls_first и nulls_last могут быть True или None, поэтому в некоторых местах вместо значений по умолчанию теперь nulls_last = None:

  - nulls_last = False
  + nulls_last = None
    if ordering_by == OrderingField.FIELD_START_AT:
        nulls_last = True

• В сигналах появился новый тип асинхронных ресиверов  (async_receivers), мы добавили его во вспомогательный класс SignalBlocker, которым временно выключаем сигналы.

• Теперь надо явно указывать USE_TZ = False, но это не проблема, мы не используем таймзоны в нашем сервисе.

• Если после refresh_from_db надо дальше по коду использовать закешированные значения из базы данных, появился такой хак:

my_model_obj._meta.private_fields = []
my_model_obj.refresh_from_db()

Новинка, которой воспользуемся в будущем, — расчетные поля по умолчанию (default) в БД. Есть сценарии, где нужно, чтобы поле по умолчанию было не скаляром, а содержало расчётное значение на основе других полей.