2024-08-08 10:38:12 +08:00

59 lines
1.5 KiB
Python

from typing import Literal
from abc import ABC, abstractmethod
from httpx import AsyncClient
from ..types import Target
from .http import http_client
class ClientManager(ABC):
@abstractmethod
async def get_client(self, target: Target | None) -> AsyncClient: ...
@abstractmethod
async def get_client_for_static(self) -> AsyncClient: ...
@abstractmethod
async def get_query_name_client(self) -> AsyncClient: ...
@abstractmethod
async def refresh_client(self): ...
class DefaultClientManager(ClientManager):
async def get_client(self, target: Target | None) -> AsyncClient:
return http_client()
async def get_client_for_static(self) -> AsyncClient:
return http_client()
async def get_query_name_client(self) -> AsyncClient:
return http_client()
async def refresh_client(self):
pass
class Site:
schedule_type: Literal["date", "interval", "cron"]
schedule_setting: dict
name: str
client_mgr: type[ClientManager] = DefaultClientManager
require_browser: bool = False
def __str__(self):
return f"[{self.name}]-{self.name}-{self.schedule_setting}"
def anonymous_site(schedule_type: Literal["date", "interval", "cron"], schedule_setting: dict) -> type[Site]:
return type(
"AnonymousSite",
(Site,),
{
"schedule_type": schedule_type,
"schedule_setting": schedule_setting,
"client_mgr": DefaultClientManager,
},
)