Watches radarr and sonarr download queues and removes downloads if they become stalled or no longer needed.
#!/usr/bin/env python
import sys
import os
import configparser

########################################################################################################################
# Check if in Docker
IS_IN_DOCKER = os.environ.get('IS_IN_DOCKER')

########################################################################################################################
def config_section_map(section):
    'Load the config file into a dictionary'
    dict1 = {}
    options = config.options(section)
    for option in options:
        try:
            dict1[option] = config.get(section, option)
        except:
            print("exception on %s!" % option)
            dict1[option] = None
    return dict1

def cast(value, type_):
    return type_(value)

def get_config_value(key, config_section, is_mandatory, datatype, default_value = None):
    'Return for each key the corresponding value from the Docker Environment or the Config File'
    if IS_IN_DOCKER:
        config_value = os.environ.get(key)
        if config_value is not None: 
            # print(f'The value retrieved for [{config_section}]: {key} is "{config_value}"')
            config_value = config_value
            # return config_value
        elif is_mandatory:
            print(f'[ ERROR ]: Variable not specified in Docker environment: {key}' )
            sys.exit(0)
        else:
            # return default_value
            config_value = default_value

    else:
        try:
            config_value = config_section_map(config_section).get(key)
        except configparser.NoSectionError:
            config_value = None
        if config_value is not None:
            # print(f'The value retrieved for [{config_section}]: {key} is "{config_value}"')
            config_value = config_value  
            # return config_value
        elif is_mandatory:
            print(f'[ ERROR ]: Mandatory variable not specified in config file, section [{config_section}]: {key} (data type: {datatype.__name__})')
            sys.exit(0)
        else:
            # return default_value 
            config_value = default_value

    # Apply data type
    try:
        if datatype == bool:
            config_value = eval(str(config_value).capitalize())
        if config_value is not None: config_value = cast(config_value, datatype)
    except:
        print(f'[ ERROR ]: The value retrieved for [{config_section}]: {key} is "{config_value}" and cannot be converted to data type {datatype}')
        sys.exit(0)
    return config_value    

########################################################################################################################
# Load Config File
config_file_name = 'config.conf'
config_file_full_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), config_file_name)
sys.tracebacklimit = 0  # dont show stack traces in prod mode
config = configparser.ConfigParser()
config.optionxform = str # maintain capitalization of config keys
config.read(config_file_full_path)

########################################################################################################################
# Load Config
# General
LOG_LEVEL                   = get_config_value('LOG_LEVEL',                     'general',      False,  str,    'INFO')
TEST_RUN                    = get_config_value('TEST_RUN',                      'general',      False,  bool,   False)

# Features
REMOVE_TIMER                = get_config_value('REMOVE_TIMER',                  'features',     False,  int,    10)
REMOVE_FAILED               = get_config_value('REMOVE_FAILED',                 'features',     False,  bool,   False)
REMOVE_STALLED              = get_config_value('REMOVE_STALLED',                'features',     False,  bool,   False)
REMOVE_METADATA_MISSING     = get_config_value('REMOVE_METADATA_MISSING',       'features',     False,  bool,   False)
REMOVE_ORPHANS              = get_config_value('REMOVE_ORPHANS' ,               'features',     False,  bool,   False)
REMOVE_UNMONITORED          = get_config_value('REMOVE_UNMONITORED' ,           'features',     False,  bool,   False)
PERMITTED_ATTEMPTS          = get_config_value('PERMITTED_ATTEMPTS',            'features',     False,  int,    3)
NO_STALLED_REMOVAL_QBIT_TAG = get_config_value('NO_STALLED_REMOVAL_QBIT_TAG',   'features',     False,  str,   'Don\'t Kill If Stalled')

# Radarr
RADARR_URL                  = get_config_value('RADARR_URL',                    'radarr',       False,  str)
RADARR_KEY                  = None if RADARR_URL == None else \
                              get_config_value('RADARR_KEY',                    'radarr',       True,   str)

# Sonarr    
SONARR_URL                  = get_config_value('SONARR_URL',                    'sonarr',       False,  str)
SONARR_KEY                  = None if SONARR_URL == None else \
                              get_config_value('SONARR_KEY',                    'sonarr',       True,   str)

# qBittorrent
QBITTORRENT_URL             = get_config_value('QBITTORRENT_URL',               'qbittorrent',  False,  str,    '')
QBITTORRENT_USERNAME        = get_config_value('QBITTORRENT_USERNAME',          'qbittorrent',  False,  str,    '')
QBITTORRENT_PASSWORD        = get_config_value('QBITTORRENT_PASSWORD',          'qbittorrent',  False,  str,    '')

########################################################################################################################
########### Validate settings
if not (RADARR_URL or SONARR_URL):
    print(f'[ ERROR ]: No Radarr/Sonarr URLs specified (nothing to monitor)')
    sys.exit(0)

########### Enrich setting variables
if RADARR_URL:      RADARR_URL      += '/api/v3'
if SONARR_URL:      SONARR_URL      += '/api/v3'
if QBITTORRENT_URL: QBITTORRENT_URL += '/api/v2'

########### Add Variables to Dictionary
settings_dict = {}
for var_name in dir():
    if var_name.isupper():
        settings_dict[var_name] = locals()[var_name]


config/config.py
ManiMatter/decluttarr
########### Import Libraries
import asyncio
import logging, verboselogs
from src.utils.rest import rest_get, rest_post
from requests.exceptions import RequestException
import json
from dateutil.relativedelta import relativedelta as rd
from config.config import settings_dict
from src.queue_cleaner import queue_cleaner
#print(json.dumps(settings_dict,indent=4))
import requests
########### Enabling Logging
# Set up logging
log_level_num=logging.getLevelName(settings_dict['LOG_LEVEL'])
logger = verboselogs.VerboseLogger(__name__)
logging.basicConfig(
    format=('' if settings_dict['IS_IN_DOCKER'] else '%(asctime)s ') + ('[%(levelname)-7s]' if settings_dict['LOG_LEVEL']=='VERBOSE' else '[%(levelname)s]') + ': %(message)s', 
    level=log_level_num 
)

class Defective_Tracker:
    # Keeps track of which downloads were already caught as stalled previously
    def __init__(self, dict):
        self.dict = dict

# Main function
async def main():
    # Get name of Radarr / Sonarr instances
    try:
        if settings_dict['RADARR_URL']:
            settings_dict['RADARR_NAME'] = (await rest_get(settings_dict['RADARR_URL']+'/system/status', settings_dict['RADARR_KEY']))['instanceName']
    except:
            settings_dict['RADARR_NAME'] = 'Radarr'
    try:
        if settings_dict['SONARR_URL']:
            settings_dict['SONARR_NAME'] = (await rest_get(settings_dict['SONARR_URL']+'/system/status', settings_dict['SONARR_KEY']))['instanceName']
    except:
        settings_dict['SONARR_NAME'] = 'Sonarr'

    # Print Settings
    fmt = '{0.days} days {0.hours} hours {0.minutes} minutes'
    logger.info('#' * 50)
    logger.info('Application Started!')
    
    logger.info('')      
    logger.info('*** Current Settings ***') 
    logger.info('%s | Removing failed downloads', str(settings_dict['REMOVE_FAILED']))
    logger.info('%s | Removing stalled downloads', str(settings_dict['REMOVE_STALLED']))
    logger.info('%s | Removing downloads missing metadata', str(settings_dict['REMOVE_METADATA_MISSING']))
    logger.info('%s | Removing orphan downloads', str(settings_dict['REMOVE_ORPHANS']))
    logger.info('%s | Removing downloads belonging to unmonitored TV shows/movies', str(settings_dict['REMOVE_UNMONITORED']))
    
    logger.info('')          
    logger.info('Running every: %s', fmt.format(rd(minutes=settings_dict['REMOVE_TIMER'])))   
    logger.info('Permitted number of times before stalled/missing metadata downloads are removed: %s', str(settings_dict['PERMITTED_ATTEMPTS']))      
    if settings_dict['QBITTORRENT_URL']: logger.info('Downloads with this tag will be skipped: \"%s\"', settings_dict['NO_STALLED_REMOVAL_QBIT_TAG'])                  
    
    logger.info('') 
    logger.info('*** Configured Instances ***')
    if settings_dict['RADARR_URL']: logger.info('%s: %s', settings_dict['RADARR_NAME'], settings_dict['RADARR_URL'])
    if settings_dict['SONARR_URL']: logger.info('%s: %s', settings_dict['SONARR_NAME'], settings_dict['SONARR_URL'])    
    if settings_dict['QBITTORRENT_URL']: logger.info('qBittorrent: %s', settings_dict['QBITTORRENT_URL'])    

    logger.info('') 
    logger.info('*** Check Instances ***')
    if settings_dict['RADARR_URL']:
        error_occured = False
        try: 
            await asyncio.get_event_loop().run_in_executor(None, lambda: requests.get(settings_dict['RADARR_URL']+'/system/status', params=None, headers={'X-Api-Key': settings_dict['RADARR_KEY']}))
            logger.info('OK | %s', settings_dict['RADARR_NAME'])
        except Exception as error:
            error_occured = True
            logger.error('-- | %s *** Error: %s ***', settings_dict['RADARR_NAME'], error)

    if settings_dict['SONARR_URL']:
        try: 
            await asyncio.get_event_loop().run_in_executor(None, lambda: requests.get(settings_dict['SONARR_URL']+'/system/status', params=None, headers={'X-Api-Key': settings_dict['SONARR_KEY']}))
            logger.info('OK | %s', settings_dict['SONARR_NAME'])
        except Exception as error:
            error_occured = True
            logger.error('-- | %s *** Error: %s ***', settings_dict['SONARR_NAME'], error)

    if settings_dict['QBITTORRENT_URL']:
        try: 
            response = await asyncio.get_event_loop().run_in_executor(None, lambda: requests.post(settings_dict['QBITTORRENT_URL']+'/auth/login', data={'username': settings_dict['QBITTORRENT_USERNAME'], 'password': settings_dict['QBITTORRENT_PASSWORD']}, headers={'content-type': 'application/x-www-form-urlencoded'}))
            if response.text == 'Fails.':
                raise ConnectionError('Login failed.')
            response.raise_for_status()
            settings_dict['QBIT_COOKIE'] = {'SID': response.cookies['SID']} 
            logger.info('OK | %s', 'qBittorrent')
        except Exception as error:
            error_occured = True
            logger.error('-- | %s *** Error: %s / Reponse: %s ***', 'qBittorrent', error, response.text)

    if error_occured:
        logger.warning('At least one instance was not reachable. Waiting for 60 seconds, then exiting Decluttarr.')      
        await asyncio.sleep(60)
        exit()

    logger.info('') 
    logger.info('#' * 50)
    if settings_dict['LOG_LEVEL'] == 'INFO':
        logger.info('LOG_LEVEL = INFO: Only logging changes (switch to VERBOSE for more info)')      
    else:
        logger.info(f'')
    if settings_dict['TEST_RUN']:
        logger.info(f'*'* 50)
        logger.info(f'*'* 50)
        logger.info(f'')
        logger.info(f'TEST_RUN FLAG IS SET!')       
        logger.info(f'THIS IS A TEST RUN AND NO UPDATES/DELETES WILL BE PERFORMED')
        logger.info(f'')
        logger.info(f'*'* 50)
        logger.info(f'*'* 50)

    # Check if Qbit Tag exists:
    if settings_dict['QBITTORRENT_URL']:
        current_tags = await rest_get(settings_dict['QBITTORRENT_URL']+'/torrents/tags',cookies=settings_dict['QBIT_COOKIE'])
        if not settings_dict['NO_STALLED_REMOVAL_QBIT_TAG'] in current_tags:
            if settings_dict['QBITTORRENT_URL']: 
                logger.info('Creating tag in qBittorrent: %s', settings_dict['NO_STALLED_REMOVAL_QBIT_TAG'])  
                if not settings_dict['TEST_RUN']:
                    await rest_post(url=settings_dict['QBITTORRENT_URL']+'/torrents/createTags', data={'tags': settings_dict['NO_STALLED_REMOVAL_QBIT_TAG']}, headers={'content-type': 'application/x-www-form-urlencoded'}, cookies=settings_dict['QBIT_COOKIE'])

    # Start application
    while True:
        logger.verbose('-' * 50)
        if settings_dict['RADARR_URL']: await queue_cleaner(settings_dict, 'radarr', defective_tracker)
        if settings_dict['SONARR_URL']: await queue_cleaner(settings_dict, 'sonarr', defective_tracker)
        logger.verbose('')  
        logger.verbose('Queue clean-up complete!')  
        await asyncio.sleep(settings_dict['REMOVE_TIMER']*60)
    return

if __name__ == '__main__':
    instances = {settings_dict['RADARR_URL']: {}} if settings_dict['RADARR_URL'] else {} + \
                {settings_dict['SONARR_URL']: {}} if settings_dict['SONARR_URL'] else {} 
    defective_tracker = Defective_Tracker(instances)
    asyncio.run(main())


main.py
ManiMatter/decluttarr
# Cleans the download queue
import logging, verboselogs
logger = verboselogs.VerboseLogger(__name__)
from src.utils.rest import (rest_get, rest_delete)
import json
from src.utils.nest_functions import (add_keys_nested_dict, nested_get)
import sys, os

class Deleted_Downloads:
    # Keeps track of which downloads have already been deleted (to not double-delete)
    def __init__(self, dict):
        self.dict = dict

async def get_queue(BASE_URL, API_KEY, params = {}):
    totalRecords = (await rest_get(f'{BASE_URL}/queue', API_KEY, params))['totalRecords']
    if totalRecords == 0:
        return None
    queue = await rest_get(f'{BASE_URL}/queue', API_KEY, {'page': '1', 'pageSize': totalRecords}|params) 
    return queue

async def remove_failed(settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads):
    # Detects failed and triggers delete. Does not add to blocklist
    queue = await get_queue(BASE_URL, API_KEY)
    if not queue: return 0
    failedItems = []    
    for queueItem in queue['records']:
        if 'errorMessage' in queueItem and 'status' in queueItem:
            if  queueItem['status']       == 'failed' or \
               (queueItem['status']       == 'warning' and queueItem['errorMessage'] == 'The download is missing files'):
                await remove_download(BASE_URL, API_KEY, queueItem['id'], queueItem['title'], queueItem['downloadId'], 'failed', False, deleted_downloads, settings_dict['TEST_RUN'])
                failedItems.append(queueItem)
    return len(failedItems)

async def remove_stalled(settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads, defective_tracker):
    # Detects stalled and triggers repeat check and subsequent delete. Adds to blocklist   
    queue = await get_queue(BASE_URL, API_KEY)
    if not queue: return 0
    if settings_dict['QBITTORRENT_URL']:
        protected_dowloadItems = await rest_get(settings_dict['QBITTORRENT_URL']+'/torrents/info',params={'tag': settings_dict['NO_STALLED_REMOVAL_QBIT_TAG']}, cookies=settings_dict['QBIT_COOKIE']  )
        protected_downloadIDs = [str.upper(item['hash']) for item in protected_dowloadItems]
    else:
        protected_downloadIDs = []
    stalledItems = []    
    for queueItem in queue['records']:
        if 'errorMessage' in queueItem and 'status' in queueItem:
            if  queueItem['status']        == 'warning' and \
                queueItem['errorMessage']  == 'The download is stalled with no connections':
                    if queueItem['downloadId'] in protected_downloadIDs:
                        logger.verbose('>>> Detected stalled download, tagged not to be killed: %s',queueItem['title'])
                    else:
                        stalledItems.append(queueItem)
    await check_permitted_attempts(settings_dict, stalledItems, 'stalled', True, deleted_downloads, BASE_URL, API_KEY, defective_tracker)
    return len(stalledItems)

async def remove_metadata_missing(settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads, defective_tracker):
    # Detects downloads stuck downloading meta data and triggers repeat check and subsequent delete. Adds to blocklist  
    queue = await get_queue(BASE_URL, API_KEY)
    if not queue: return 0
    missing_metadataItems = []
    for queueItem in queue['records']:
        if 'errorMessage' in queueItem and 'status' in queueItem:
            if  queueItem['status']        == 'queued' and \
                queueItem['errorMessage']  == 'qBittorrent is downloading metadata':
                    missing_metadataItems.append(queueItem)
    await check_permitted_attempts(settings_dict, missing_metadataItems, 'missing metadata', True, deleted_downloads, BASE_URL, API_KEY, defective_tracker)
    return len(missing_metadataItems)

async def remove_orphans(settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads):
    # Removes downloads belonging to movies/tv shows that have been deleted in the meantime
    full_queue = await get_queue(BASE_URL, API_KEY, params = {'includeUnknownMovieItems' if radarr_or_sonarr == 'radarr' else 'includeUnknownSeriesItems': 'true'})
    if not full_queue: return 0 # By now the queue may be empty 
    queue = await get_queue(BASE_URL, API_KEY) 
    full_queue_items = [{'id': queueItem['id'], 'title': queueItem['title'], 'downloadId': queueItem['downloadId']} for queueItem in full_queue['records']]
    if queue:
        queue_ids = [queueItem['id'] for queueItem in queue['records']]
    else:
        queue_ids = []
    orphanItems = [{'id': queueItem['id'], 'title': queueItem['title'], 'downloadId': queueItem['downloadId']} for queueItem in full_queue_items if queueItem['id'] not in queue_ids]
    for queueItem in orphanItems:
        await remove_download(settings_dict, BASE_URL, API_KEY, queueItem['id'], queueItem['title'], queueItem['downloadId'], 'orphan', False, deleted_downloads)
    return len(orphanItems)

async def remove_unmonitored(settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads):
    # Removes downloads belonging to movies/tv shows that are not monitored
    queue = await get_queue(BASE_URL, API_KEY) 
    if not queue: return 0
    unmonitoredItems= []
    downloadItems = []
    for queueItem in queue['records']:
        if radarr_or_sonarr == 'sonarr': 
            monitored = (await rest_get(f'{BASE_URL}/episode/{str(queueItem["episodeId"])}', API_KEY))['monitored']
        else: 
            monitored = (await rest_get(f'{BASE_URL}/movie/{str(queueItem["movieId"])}', API_KEY))['monitored']
        downloadItems.append({'downloadId': queueItem['downloadId'], 'id': queueItem['id'], 'monitored': monitored})
    monitored_downloadIds = [downloadItem['downloadId'] for downloadItem in downloadItems if downloadItem['monitored']]
    unmonitoredItems = [downloadItem for downloadItem in downloadItems if downloadItem['downloadId'] not in monitored_downloadIds]
    for unmonitoredItem in unmonitoredItems:
        await remove_download(settings_dict, BASE_URL, API_KEY, queueItem['id'], queueItem['title'], queueItem['downloadId'], 'unmonitored', False, deleted_downloads)
    return len(unmonitoredItems)

async def check_permitted_attempts(settings_dict, current_defective_items, failType, blocklist, deleted_downloads, BASE_URL, API_KEY, defective_tracker):
    # Checks if downloads are repeatedly found as stalled / stuck in metadata and if yes, deletes them
    # 1. Create list of currently defective
    current_defective = {}
    for queueItem in current_defective_items:
        current_defective[queueItem['id']] = {'title': queueItem['title'],'downloadId': queueItem['downloadId']}
        
    # 2. Check if those that were previously defective are no longer defective -> those are recovered
    try:
        recovered_ids = [tracked_id for tracked_id in defective_tracker.dict[BASE_URL][failType] if tracked_id not in current_defective]
    except KeyError:
        recovered_ids = []
    for recovered_id in recovered_ids:
       del defective_tracker.dict[BASE_URL][failType][recovered_id]
    # 3. For those that are defective, add attempt + 1 if present before, or make attempt = 0. If exceeding number of permitted attempts, delete hem
    download_ids_stuck = []
    for queueId in current_defective:
        try: 
            defective_tracker.dict[BASE_URL][failType][queueId]['Attempts'] += 1
        except KeyError:
            await add_keys_nested_dict(defective_tracker.dict,[BASE_URL, failType, queueId], {'title': current_defective[queueId]['title'], 'downloadId': current_defective[queueId]['downloadId'], 'Attempts': 1})
        if current_defective[queueId]['downloadId'] not in download_ids_stuck:
            download_ids_stuck.append(current_defective[queueId]['downloadId'])
            logger.info('>>> Detected %s download (%s out of %s permitted times): %s', failType, str(defective_tracker.dict[BASE_URL][failType][queueId]['Attempts']), str(settings_dict['PERMITTED_ATTEMPTS']), defective_tracker.dict[BASE_URL][failType][queueId]['title'])
        if defective_tracker.dict[BASE_URL][failType][queueId]['Attempts'] > settings_dict['PERMITTED_ATTEMPTS']:
            await remove_download(settings_dict, BASE_URL, API_KEY, queueId, current_defective[queueId]['title'], current_defective[queueId]['downloadId'],  failType, blocklist, deleted_downloads)
    return

async def remove_download(settings_dict, BASE_URL, API_KEY, queueId, queueTitle, downloadId, failType, blocklist, deleted_downloads):
    # Removes downloads and creates log entry
    if downloadId not in deleted_downloads.dict:
        logger.info('>>> Removing %s download: %s', failType, queueTitle)
        if not settings_dict['TEST_RUN']: await rest_delete(f'{BASE_URL}/queue/{queueId}', API_KEY, {'removeFromClient': 'true', 'blocklist': blocklist}) 
        deleted_downloads.dict.append(downloadId)    
    return

########### MAIN FUNCTION ###########
async def queue_cleaner(settings_dict, radarr_or_sonarr, defective_tracker):
    # Read out correct instance depending on radarr/sonarr flag
    run_dict = {}
    if radarr_or_sonarr == 'radarr':
        BASE_URL    = settings_dict['RADARR_URL']
        API_KEY     = settings_dict['RADARR_KEY']
        NAME        = settings_dict['RADARR_NAME']
    else:
        BASE_URL    = settings_dict['SONARR_URL']
        API_KEY     = settings_dict['SONARR_KEY']
        NAME        = settings_dict['SONARR_NAME']

    # Cleans up the downloads queue
    logger.verbose('Cleaning queue on %s:', NAME)
    try:
        full_queue = await get_queue(BASE_URL, API_KEY, params = {'includeUnknownMovieItems' if radarr_or_sonarr == 'radarr' else 'includeUnknownSeriesItems': 'true'})
        if not full_queue: 
            logger.verbose('>>> Queue is empty.')
            return
              
        deleted_downloads = Deleted_Downloads([])
        items_detected = 0
        if settings_dict['REMOVE_FAILED']:
            items_detected += await remove_failed(            settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads)
        
        if settings_dict['REMOVE_STALLED']: 
            items_detected += await remove_stalled(           settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads, defective_tracker)

        if settings_dict['REMOVE_METADATA_MISSING']: 
            items_detected += await remove_metadata_missing(  settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads, defective_tracker)

        if settings_dict['REMOVE_ORPHANS']: 
            items_detected += await remove_orphans(           settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads)

        if settings_dict['REMOVE_UNMONITORED']: 
            items_detected += await remove_unmonitored(       settings_dict, radarr_or_sonarr, BASE_URL, API_KEY, deleted_downloads)

        if items_detected == 0:
            logger.verbose('>>> Queue is clean.')
    except Exception as error:
            exc_type, exc_obj, exc_tb = sys.exc_info()
            fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
            logger.warning('>>> Queue cleaning failed on %s. (File: %s / Line: %s / Error Message: %s / Error Type: %s)', NAME, fname, exc_tb.tb_lineno, error, exc_type)




src/queue_cleaner.py
ManiMatter/decluttarr

async def nested_set(dic, keys, value, matchConditions=None):
    # Sets the value of a key in a dictionary to a certain value. 
    # If multiple items are present, it can filter for a matching item
    for key in keys[:-1]:
        dic = dic.setdefault(key, {})
    if matchConditions:
        i = 0
        match = False
        for item in dic:   
            for matchCondition in matchConditions:
                if item[matchCondition] != matchConditions[matchCondition]:
                    match = False
                    break
                else:
                    match = True
            if match:
                dic = dic[i]
                break
            i += 1
    dic[keys[-1]] = value
    

async def add_keys_nested_dict(d, keys, defaultValue = None):
    # Creates a nested value if key does not exist
    for key in keys[:-1]:
        if key not in d:
            d[key] = {}
        d = d[key]
    d.setdefault(keys[-1], defaultValue)

async def nested_get(dic, return_attribute, matchConditions):
    # Retrieves a list contained in return_attribute, found within dic based on matchConditions
    i = 0
    match = False
    hits = []
    for item in dic:   
        for matchCondition in matchConditions:
            if item[matchCondition] != matchConditions[matchCondition]:
                match = False
                break
            else:
                match = True
        if match:
            hits.append(dic[i][return_attribute])
        i += 1
    return hits
src/utils/nest_functions.py
ManiMatter/decluttarr
########### Functions to call radarr/sonarr APIs
import logging
import asyncio
import requests
from requests.exceptions import RequestException
import json
from config.config import settings_dict

# GET
async def rest_get(url, api_key=None, params=None, cookies=None):
    try:
        headers = {'X-Api-Key': api_key} if api_key else None 
        response = await asyncio.get_event_loop().run_in_executor(None, lambda: requests.get(url, params=params, headers=headers, cookies=cookies))
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        logging.error(f'Error making API request to {url}: {e}')
        return None
    except ValueError as e:
        logging.error(f'Error parsing JSON response from {url}: {e}')
        return None

# DELETE
async def rest_delete(url, api_key, params=None):
    if settings_dict['TEST_RUN']: return
    try:
        headers = {'X-Api-Key': api_key}
        response = await asyncio.get_event_loop().run_in_executor(None, lambda: requests.delete(url, params=params, headers=headers))
        response.raise_for_status()
        if response.status_code in [200, 204]:
            return None
        return response.json()
    except RequestException as e:
        logging.error(f'Error making API request to {url}: {e}')
        return None
    except ValueError as e:
        logging.error(f'Error parsing JSON response from {url}: {e}')
        return None

# POST
async def rest_post(url, data, headers, cookies=None):
    if settings_dict['TEST_RUN']: return
    try:
        response = await asyncio.get_event_loop().run_in_executor(None, lambda: requests.post(url, data=data, headers=headers, cookies=cookies))
        response.raise_for_status()
        if response.status_code in (200,201):
            return None
        return response.json()
    except RequestException as e:
        logging.error(f'Error making API request to {url}: {e}')
        return None
    except ValueError as e:
        logging.error(f'Error parsing JSON response from {url}: {e}')
        return None


# PUT
async def rest_put(url, api_key, data):
    if settings_dict['TEST_RUN']: return
    try:
        headers = {'X-Api-Key': api_key} | {"content-type": "application/json"}
        response = await asyncio.get_event_loop().run_in_executor(None, lambda: requests.put(url, data=data, headers=headers))
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        logging.error(f'Error making API request to {url}: {e}')
        return None
    except ValueError as e:
        logging.error(f'Error parsing JSON response from {url}: {e}')
        return None


src/utils/rest.py
ManiMatter/decluttarr
import pytest
from src.utils.nest_functions import nested_set, add_keys_nested_dict, nested_get
import asyncio

# Dictionary that is modified / queried as part of tests
input_dict =              { 1: {'name': 'Breaking Bad 1', 'data': {'episodes': 3, 'year': 1991, 'actors': ['Peter', 'Paul', 'Ppacey']}}, 
                            2: {'name': 'Breaking Bad 2', 'data': {'episodes': 6, 'year': 1992, 'actors': ['Weter', 'Waul', 'Wpacey']}},
                            3: {'name': 'Breaking Bad 3', 'data': {'episodes': 9, 'year': 1993, 'actors': ['Zeter', 'Zaul', 'Zpacey']}}}

@pytest.mark.asyncio
async def test_nested_set():
    expected_output    =  { 1: {'name': 'Breaking Bad 1', 'data': {'episodes': 3, 'year': 1991, 'actors': ['Peter', 'Paul', 'Ppacey']}}, 
                            2: {'name': 'Breaking Bad 2', 'data': {'episodes': 6, 'year': 1994, 'actors': ['Weter', 'Waul', 'Wpacey']}},
                            3: {'name': 'Breaking Bad 3', 'data': {'episodes': 9, 'year': 1993, 'actors': ['Zeter', 'Zaul', 'Zpacey']}}}
    output = input_dict
    await nested_set(output, [2, 'data' ,'year'], 1994)
    assert expected_output == output

@pytest.mark.asyncio
async def test_nested_set_conditions():
    input               = { 1: [{'year': 2001, 'rating': 'high'}, {'year': 2002, 'rating': 'high'}, {'year': 2003, 'rating': 'high'}],
                            2: [{'year': 2001, 'rating': 'high'}, {'year': 2002, 'rating': 'high'}, {'year': 2003, 'rating': 'high'}]}
    expected_output     = { 1: [{'year': 2001, 'rating': 'high'}, {'year': 2002, 'rating': 'high'}, {'year': 2003, 'rating': 'high'}],
                            2: [{'year': 2001, 'rating': 'high'}, {'year': 2002, 'rating': 'high'}, {'year': 2003, 'rating': 'LOW'}]}
    output = input
    await nested_set(output, [2, 'rating'], 'LOW', {'year': 2003})
    assert expected_output == output

@pytest.mark.asyncio
async def test_nested_set_conditions_multiple():
    input               = { 1: [{'rating': 'high', 'color': 1, 'stack': 1}, {'rating': 'high', 'color': 2, 'stack': 2}, {'rating': 'high', 'color': 2, 'stack': 1}]}
    expected_output     = { 1: [{'rating': 'high', 'color': 1, 'stack': 1}, {'rating': 'high', 'color': 2, 'stack': 2}, {'rating': 'LOW', 'color': 2, 'stack': 1}]}
    output = input
    await nested_set(output, [1, 'rating'], 'LOW', {'color': 2, 'stack': 1})
    assert expected_output == output

@pytest.mark.asyncio
async def test_add_keys_nested_dict():
    expected_output    =  { 1: {'name': 'Breaking Bad 1', 'data': {'episodes': 3, 'year': 1991, 'actors': ['Peter', 'Paul', 'Ppacey']}}, 
                            2: {'name': 'Breaking Bad 2', 'data': {'episodes': 6, 'year': 1994, 'actors': ['Weter', 'Waul', 'Wpacey'], 'spaceship': True}},
                            3: {'name': 'Breaking Bad 3', 'data': {'episodes': 9, 'year': 1993, 'actors': ['Zeter', 'Zaul', 'Zpacey']}}}
    output = input_dict
    await add_keys_nested_dict(output, [2, 'data' ,'spaceship'], True)
    assert expected_output == output


@pytest.mark.asyncio
async def test_nested_get():
    input               = { 1: [{'name': 'A', 'color': 1, 'stack': 1}, {'name': 'B', 'color': 2, 'stack': 2}, {'name': 'C', 'color': 2, 'stack': 1}]}
    expected_output     = ['C']
    output = await nested_get(input[1], 'name', {'color': 2, 'stack': 1})
    assert expected_output == output


tests/test_nest_functions.py
ManiMatter/decluttarr