"""Functions to perform common checks. These are useful for reusability, and hence maintainability of the code. """ import logging from typing import Callable from functools import wraps from gn_libs.mysqldb import Connection, database_connection from flask import flash, url_for, redirect, current_app as app from uploader.species.models import species_by_id from uploader.population.models import population_by_species_and_id logger = logging.getLogger(__name__) def with_species(redirect_uri: str): """Ensure the species actually exists.""" def __decorator__(function): @wraps(function) def __with_species__(**kwargs): try: species_id = int(kwargs.get("species_id")) if not bool(species_id): flash("Expected species_id value to be present!", "alert-danger") return redirect(url_for(redirect_uri)) with database_connection(app.config["SQL_URI"]) as conn: species = species_by_id(conn, species_id) if not bool(species): flash("Could not find species with that ID", "alert-danger") return redirect(url_for(redirect_uri)) except ValueError as _verr: logger.debug( "Exception converting value to integer: %s", kwargs.get("species_id"), exc_info=True) flash("Expected an integer for 'species_id' value.", "alert-danger") return redirect(url_for(redirect_uri)) return function(**{**kwargs, "species": species}) return __with_species__ return __decorator__ def with_population(species_redirect_uri: str, redirect_uri: str): """Ensure the population actually exists.""" def __decorator__(function): @wraps(function) @with_species(redirect_uri=species_redirect_uri) def __with_population__(**kwargs): try: species_id = int(kwargs["species_id"]) population_id = int(kwargs.get("population_id")) select_population_uri = redirect(url_for( redirect_uri, species_id=species_id)) if not bool(population_id): flash("Expected population_id value to be present!", "alert-danger") return select_population_uri with database_connection(app.config["SQL_URI"]) as conn: population = population_by_species_and_id( conn, species_id, population_id) if not bool(population): flash("Could not find population with that ID", "alert-danger") return select_population_uri except ValueError as _verr: logger.debug( "Exception converting value to integer: %s", kwargs.get("population_id"), exc_info=True) flash("Expected an integer for 'population_id' value.", "alert-danger") return select_population_uri return function(**{**kwargs, "population": population}) return __with_population__ return __decorator__ def with_dataset( species_redirect_uri: str, population_redirect_uri: str, redirect_uri: str, dataset_by_id: Callable[ [Connection, int, int, int], dict] ): """Ensure the dataset actually exists.""" def __decorator__(func): @wraps(func) @with_population(species_redirect_uri, population_redirect_uri) def __with_dataset__(**kwargs): try: _spcid = int(kwargs["species_id"]) _popid = int(kwargs["population_id"]) _dsetid = int(kwargs.get("dataset_id")) select_dataset_uri = redirect(url_for( redirect_uri, species_id=_spcid, population_id=_popid)) if not bool(_dsetid): flash("You need to select a valid 'dataset_id' value.", "alert-danger") return select_dataset_uri with database_connection(app.config["SQL_URI"]) as conn: dataset = dataset_by_id(conn, _spcid, _popid, _dsetid) if not bool(dataset): flash("You must select a valid dataset.", "alert-danger") return select_dataset_uri except ValueError as _verr: logger.debug( "Exception converting 'dataset_id' to integer: %s", kwargs.get("dataset_id"), exc_info=True) flash("Expected 'dataset_id' value to be an integer." "alert-danger") return select_dataset_uri return func(**{**kwargs, "dataset": dataset}) return __with_dataset__ return __decorator__