115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
from datetime import datetime, timedelta
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
import sys
|
|
import time
|
|
from typing import Dict, Any
|
|
|
|
import yaml
|
|
|
|
# pylint: disable=relative-beyond-top-level
|
|
from ..api_client import APIclient
|
|
from ..db import DB
|
|
from ..const import DB_SCHEMA_VER
|
|
from .extractor import Extractor
|
|
from ..models.extractor_config import ExtractorConfig
|
|
from ..models.db_config import DbConfig
|
|
|
|
|
|
def __get_config(yml_path: str) -> ExtractorConfig:
|
|
"""
|
|
get_config loads yml file and generates instance
|
|
"""
|
|
with open(yml_path, 'r', encoding='utf-8') as flr:
|
|
raw = yaml.safe_load(flr)
|
|
|
|
# adding new attrs
|
|
raw['oldest_build_age'] = datetime.now().astimezone() \
|
|
- timedelta(days=raw['data_store_days'])
|
|
|
|
# Dbconfig
|
|
db_params: Dict[str, Any] = {'name': raw['db_name'],
|
|
'username': raw['db_username'],
|
|
'password': raw['db_password'], }
|
|
if 'db_port' in raw:
|
|
db_params['port'] = raw['db_port']
|
|
if 'db_host' in raw:
|
|
db_params['host'] = raw['db_host']
|
|
raw['db_config'] = DbConfig(**db_params)
|
|
|
|
if 'oldest_to_update_days' in raw:
|
|
raw['oldest_to_update_days'] = datetime.now().astimezone() \
|
|
- timedelta(days=raw['oldest_to_update_days'])
|
|
|
|
return ExtractorConfig(**raw)
|
|
|
|
|
|
def start(yml_path: str):
|
|
config = __get_config(yml_path)
|
|
|
|
# configuring logging
|
|
logging.basicConfig(level=logging.INFO,
|
|
format='%(asctime)s %(levelname)s %(funcName)s() %(message)s',
|
|
handlers=[RotatingFileHandler(config.log_file,
|
|
maxBytes=10000000,
|
|
backupCount=3)])
|
|
|
|
# some pre-flight checks
|
|
db = DB(config.db_config)
|
|
cur_version = db.get_db_schema_version()
|
|
if not cur_version:
|
|
logging.error(
|
|
'Cant get db schema version. Make sure that schema_version exists')
|
|
sys.exit(1)
|
|
if cur_version != DB_SCHEMA_VER:
|
|
logging.error('unsupported DB schema: want %s, have %s',
|
|
DB_SCHEMA_VER, cur_version)
|
|
sys.exit(1)
|
|
db.close_conn()
|
|
|
|
while True:
|
|
logging.info('Starting extraction proccess')
|
|
api = APIclient(api_root=config.albs_url,
|
|
jwt=config.jwt, timeout=config.api_timeout)
|
|
db = DB(config.db_config)
|
|
extractor = Extractor(config, api, db)
|
|
|
|
logging.info('Starting builds insertion')
|
|
inserted_count = -1
|
|
try:
|
|
inserted_count = extractor.extract_and_store()
|
|
except Exception as err: # pylint: disable=broad-except
|
|
logging.critical("Unhandled exception %s", err, exc_info=True)
|
|
else:
|
|
logging.info(
|
|
'Build extaction was finished. %d builds were inserted', inserted_count)
|
|
|
|
logging.info('starting old builds removal')
|
|
try:
|
|
extractor.build_cleanup()
|
|
except Exception as err: # pylint: disable=broad-except
|
|
logging.critical("unhandled exception %s", err, exc_info=True)
|
|
else:
|
|
logging.info('cleanup finished')
|
|
|
|
logging.info('updating statuses of unfinished build tasks')
|
|
try:
|
|
extractor.update_builds()
|
|
except Exception as err: # pylint: disable=broad-except
|
|
logging.critical("unhandled exception %s", err, exc_info=True)
|
|
else:
|
|
logging.info('update finished')
|
|
|
|
logging.info('updating/inserting test tasks')
|
|
try:
|
|
extractor.updating_test_tasks()
|
|
except Exception as err: # pylint: disable=broad-except
|
|
logging.critical("unhandled exception %s", err, exc_info=True)
|
|
else:
|
|
logging.info('test tasks were updated')
|
|
|
|
extractor.db.close_conn()
|
|
logging.info("Extraction was finished")
|
|
logging.info("Sleeping for %d seconds", config.scrape_interval)
|
|
time.sleep(config.scrape_interval)
|