jsdio/ips_etl
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Вынес часть настроек в MiscSettings

Browse Source
This commit is contained in:
Григорич 2023-06-06 14:15:04 +03:00
parent 81dc7a1527
commit d3432ebd25
3 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/extractor.py
View File

@ -8,13 +8,14 @@ import backoff
from environs import load_dotenv from environs import load_dotenv
from models import Abitr from models import Abitr
from settings import ApiConfig, EtlConfig from settings import ApiConfig, EtlConfig, MiscSettings
from state import State from state import State
load_dotenv() load_dotenv()
etl_config = EtlConfig() etl_config = EtlConfig()
misc_settings = MiscSettings()
class ApiExtractor: class ApiExtractor:
@ -32,8 +33,9 @@ class ApiExtractor:
data=data) as resp: data=data) as resp:
return json.loads(await resp.text()) return json.loads(await resp.text())
@backoff.on_exception(backoff.expo, (aiohttp.ClientResponseError, aiohttp.ClientConnectorError), base=2, factor=1, @backoff.on_exception(backoff.expo, (aiohttp.ClientResponseError, aiohttp.ClientConnectorError,
max_value=int(etl_config.max_wait_size), max_tries=None) aiohttp.ServerDisconnectedError), base=2, factor=1,
max_value=int(misc_settings.max_wait_size), max_tries=None)
async def get_extract_data(self, state: State, iblock_id: int, fields: list = None, **kwargs) -> list[Abitr]: async def get_extract_data(self, state: State, iblock_id: int, fields: list = None, **kwargs) -> list[Abitr]:
data = { data = {

9
src/loader.py
View File

@ -8,11 +8,12 @@ import backoff as backoff
from environs import load_dotenv from environs import load_dotenv
from models import Abitr from models import Abitr
from settings import ApiConfig, EtlConfig from settings import ApiConfig, EtlConfig, MiscSettings
from state import State, BaseStorage from state import State, BaseStorage
load_dotenv() load_dotenv()
etl_config = EtlConfig() etl_config = EtlConfig()
misc_settings = MiscSettings()
class EtlLoader: class EtlLoader:
@ -21,8 +22,8 @@ class EtlLoader:
self.etl = EtlConfig() self.etl = EtlConfig()
self.api_config = ApiConfig() self.api_config = ApiConfig()
@backoff.on_exception(backoff.expo, (asyncio.TimeoutError, aiohttp.ClientConnectorError), base=2, factor=1, @backoff.on_exception(backoff.expo, (asyncio.TimeoutError, aiohttp.ClientConnectorError,), base=2, factor=1,
max_value=int(etl_config.max_wait_size), max_tries=None) max_value=int(misc_settings.max_wait_size), max_tries=None)
async def load_data(self, state: State, abitr: Abitr, iblock_id: int, storage: BaseStorage): async def load_data(self, state: State, abitr: Abitr, iblock_id: int, storage: BaseStorage):
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
logging.info(f"Информация об абитуриенте: {abitr.FIO}") logging.info(f"Информация об абитуриенте: {abitr.FIO}")
@ -33,7 +34,7 @@ class EtlLoader:
async with session.post( async with session.post(
url=f"{self.etl.protocol}://{self.etl.host}:{self.etl.port}", url=f"{self.etl.protocol}://{self.etl.host}:{self.etl.port}",
data=data_dict, data=data_dict,
timeout=aiohttp.ClientTimeout(total=int(etl_config.max_wait_size)) timeout=aiohttp.ClientTimeout(total=int(misc_settings.max_wait_size))
) as response: ) as response:
pass pass
state.set_state(f'iblock_{iblock_id}', abitr.ID) state.set_state(f'iblock_{iblock_id}', abitr.ID)

4
src/settings.py
View File

@ -18,4 +18,8 @@ class EtlConfig:
host: str = os.environ.get('DESTINATION_HOST') host: str = os.environ.get('DESTINATION_HOST')
port: int = int(os.environ.get('DESTINATION_PORT')) port: int = int(os.environ.get('DESTINATION_PORT'))
protocol: str = os.environ.get('DESTINATION_PROTOCOL') protocol: str = os.environ.get('DESTINATION_PROTOCOL')
@dataclass
class MiscSettings:
max_wait_size: int = os.environ.get('MAX_WAIT_SIZE', 60) max_wait_size: int = os.environ.get('MAX_WAIT_SIZE', 60)
use_notify: bool = os.environ.get('USE_NOTIFY', False) == 'True'