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

Данные повторно отправляться не будут, если доставка была с ошибкой

Browse Source
This commit is contained in:
Григорич 2023-06-27 12:29:32 +03:00
parent 76fafc90e1
commit 9121013eb8
2 changed files with 10 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/extractor.py
View File

@ -1,7 +1,6 @@
import logging import logging
import json import json
import os import os
import time
import aiohttp import aiohttp
import backoff import backoff

22
src/loader.py
View File

@ -1,10 +1,8 @@
import asyncio import asyncio
import json import json
import logging import logging
import time
import aiohttp import aiohttp
import backoff as backoff
from environs import load_dotenv from environs import load_dotenv
from models import Abitr from models import Abitr
@ -22,9 +20,6 @@ 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,
aiohttp.ServerDisconnectedError), base=2, factor=1,
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}")
@ -32,14 +27,17 @@ class EtlLoader:
data_dict.pop('ID') data_dict.pop('ID')
logging.info('Отправка данных к принимающей стороне... ') logging.info('Отправка данных к принимающей стороне... ')
logging.info(f'Данные: \n{json.dumps(data_dict, ensure_ascii=False)}') logging.info(f'Данные: \n{json.dumps(data_dict, ensure_ascii=False)}')
async with session.post( try:
url=f"{self.etl.protocol}://{self.etl.host}:{self.etl.port}", async with session.post(
data=data_dict, url=f"{self.etl.protocol}://{self.etl.host}:{self.etl.port}",
timeout=aiohttp.ClientTimeout(total=int(misc_settings.max_wait_size)) data=data_dict,
) as response: timeout=aiohttp.ClientTimeout(total=int(misc_settings.max_wait_size))
pass ) as response:
logging.info('Данные успешно доставлены!')
except (asyncio.TimeoutError, aiohttp.ClientConnectorError, aiohttp.ServerDisconnectedError):
logging.error('Данные доставлены с ошибкой')
state.set_state(f'iblock_{iblock_id}', abitr.ID) state.set_state(f'iblock_{iblock_id}', abitr.ID)
storage.save_state(state.local_state) storage.save_state(state.local_state)
logging.info('Данные успешно доставлены!')
await asyncio.sleep(1) await asyncio.sleep(1)