planning of architecture

tasks.py

from decimal import Decimal

import pydantic


@shared_task
def get_provider_api_stocks(api_key, url_constructor, deserialize_constructor):
    # api_key в принципе нельзя передавать так
    return True


from typing import Protocol


class BaseCodeProviderClient(Protocol):

    @abstractmethod
    def get_one_game_detail_stock():
        return True

    @abstractmethod
    def get_codes_from_order():
        return True

    @abstractmethod
    def get_orders_history():
        return True

    @abstractmethod
    def get_detail_of_order(self, order_id: str):
        return True


def get_profitable_stocks_info(params):
    get_provider_api_stocks(params)
    get_one_game_detail_stock(params)


class HTTPClient:

    def __init__(
        self,
        base_url: str,
        retry_policy: Retry(connect=2, read=2),
        timeout_policy: TimePolicy(5, 20),
	    auth: BasicHTTPAuth(user, password),
        middleware = [HTTPLoggingMiddleware, HTTPMetricsMiddleware, HTTPXRequestID],
        session: Optional[requests.Session] = None  # requests.Session(),
    ):
        pass

class OrderDetails(pydantic.BaseModel):
    id: int
    amount: Decimal
    code: str


class GameEXECodesProvider(BaseCodeProviderClient):
    slug = 'game_exe'

    def __init__(self, provider: 'CodeProvider'):

        self._provider = provider
        self._http_client = HTTPClient(
            base_url=provider.url,
            auth=provider.get_auth(),
        )

    def get_detail_of_order(self, order_id: str) -> OrderDetails:
        response = self._http_client.get(f'/orders/{order_id}')
        return OrderDetails.parse_obj(response.json())



class Codes1Provider(BaseCodeProviderClient):
    slug = 'code_1'


    def get_one_game_detail_stock(self):
        """do something"""


class Codes2Provider(BaseCodeProviderClient):
    slug = 'code_2'


    def get_orders_history(self, ...):
        """do something"""


AVAILABLE_PROVIDERS = {
    GameEXECodesProvider.slug: GameEXECodesProvider,
    Codes1Provider.slug: Codes1Provider,
    Codes2Provider.slug: Codes2Provider,
}


class CodeProvider(Model):
    slug = TextField(unique=True)
    url = URLField()

    # Authentication data can be different between providers
    auth_data = JSONField()



@shared_task
def gather_new_codes_dispatcher():
    """Scheduled tasks, executed every hour and gather all new available codes from providers."""
    for provider_id in CodeProvider.objects.values_list('id', flat=True):
        gather_new_cored_for_provider(provider_id=provider_id).apply_async()


@shared_task
def gather_new_cored_for_provider(*, provider_id: int):
    provider = CodeProvider.objects.get(id=provider_id)
    provider_client: BaseCodeProviderClient = AVAILABLE_PROVIDERS[provider.slug](provider=provider)
    codes = provider_client.gather_codes()
    save_codes()

И что? Что с ним может произойти?

попадет в логи селери, потом эти логи кто-нибудь случайно получит и будет знать ключ.

И что? Что с ним может произойти?

попадет в логи селери, потом эти логи кто-нибудь случайно получит и будет знать ключ.

а если его прокидывать в объект класса, то он не будет в логах селери что ли?

Покажи мне как эту всю конструкцию ты в итоге предлагаешь вызывать/использовать?

И, возможно, это было не очень понятно, но все эти параметры потенциально содержатся в каждой из тех функций, что я перечислил api_key, url_constructor, deserialize_constructor

BaseCodeProvider(Protocol):

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

Кстати говоря, не уверен, но по моему в аннотациях можно проверить на тип класса и без его наследования от протоколов.

а если его прокидывать в объект класса, то он не будет в логах селери что ли?

а объект класса тоже не стоит прокидывать

Покажи мне как эту всю конструкцию ты в итоге предлагаешь вызывать/использовать?

Дополнил гист

return OrderDetails.parse_obj(response.json()) — ухты, кайф. Выглядит как что-то, что я давно искал. А более сложные данные с более глубоким уровнем вложенноти он может десерелиазовать? Типо словарь со словарями словарей и списков?

Codes1Provider и второй же не будут работать, т.к. недостаточно переопределено методов, верно?

это не рабочий код, это пример, все методы должны быть определены

GameEXECodesProvider — а что ты имел ввиду под exe? .exe расширение или что провайдер бывший? или ещё что-то?

я придумал это имя

я придумал это имя

ну это понятно. Какой ты смысл вкладывал, я имею ввиду?

а почему нельзя прямо в коде во время инициализации объекта доставать его из базы? Зачем его явно передавать?

def __init__(self):

        self._provider = Provider.objects.get(slug=self.slug)
        self._http_client = HTTPClient(
            base_url=self._provider.url,
            auth=self._provider.get_auth(),
        )

в __init__ не стоит вообще никаких сложных действий делать (ходить в базу, ходить по http) и т.д. так как это считается сайд эффектом, в инициализаторе обычно идет присвоение данных, сами данные передаются из получаются из вне.

В указанном тобой случае класс бы выглядел так:

class GameEXECodesProvider(BaseCodeProviderClient):

    def __init__(self, provider_slug: str):
        self._provider_slug = provider_slug

    @cached_property
    def provider(self) -> 'Provider':
        return Provider.objects.get(slug=self.slug)

в __init__ не стоит вообще никаких сложных действий делать (ходить в базу, ходить по http) и т.д. так как это считается сайд эффектом, в инициализаторе обычно идет присвоение данных, сами данные передаются из получаются из вне.

В указанном тобой случае класс бы выглядел так:

class GameEXECodesProvider(BaseCodeProviderClient):

    def __init__(self, provider_slug: str):
        self._provider_slug = provider_slug

    @cached_property
    def provider(self) -> 'Provider':
        return Provider.objects.get(slug=self.slug)

аа, ну да. Не вопрос. Так, что такой вариант c получением объекта из класса, тоже норм ?