Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def add_run(session, method, cmdline, date, status, name):
"""Create a new Run and add it to the session.
:param session: live SQLAlchemy session of pyani database
:param method: string describing analysis run type
:param cmdline: string describing pyani command-line for run
:param date: datetime object describing analysis start time
:param status: string describing status of analysis
:param name: string - name given to the analysis run
Creates a new Run object with the passed parameters, and returns it.
"""
try:
run = Run(method=method, cmdline=cmdline, date=date, status=status, name=name)
except Exception:
raise PyaniORMException(
f"Could not create {method} run with command line: {cmdline}"
)
try:
session.add(run)
session.commit()
except Exception:
raise PyaniORMException(f"Could not add run {run} to the database")
return run
"""
logger = logging.getLogger(__name__)
# Output formats will apply across all tabular data requested
# Expect comma-separated format arguments, and turn them into an iterable
formats = process_formats(args)
logger.info(termcolor("Creating report output in formats: %s", "red"), formats)
# Declare which database is being used, and connect to session
logger.debug("Using database: %s", args.dbpath)
session = pyani_orm.get_session(args.dbpath)
# Report runs in the database
if args.show_runs:
statement = session.query(
Run.run_id, Run.name, Run.method, Run.date, Run.cmdline
).statement
headers = ["run ID", "name", "method", "date run", "command-line"]
report(args, session, formats, ReportParams("runs", statement, headers))
# Report genomes in the database
if args.show_genomes:
statement = session.query(
Genome.genome_id,
Genome.description,
Genome.path,
Genome.genome_hash,
Genome.length,
).statement
headers = ["genome ID", "description", "path", "MD5 hash", "genome length"]
report(args, session, formats, ReportParams("genomes", statement, headers))
def get_matrix_labels_for_run(session: Any, run_id: int) -> Dict:
"""Return dictionary of genome labels, keyed by row/column ID.
:param session: live SQLAlchemy session
:param run_id: the Run.run_id value for matrices
The labels should be valid for identity, coverage and other complete
matrix results accessed via the .df_* attributes of a run.
Labels are returned keyed by the string of the genome ID, for compatibility with
matplotlib.
"""
results = (
session.query(Genome.genome_id, Label.label)
.join(rungenome, Run)
.join(
Label, and_(Genome.genome_id == Label.genome_id, Run.run_id == Label.run_id)
)
.filter(Run.run_id == run_id)
.all()
)
return {str(_.genome_id): _.label for _ in results}
)
# Report table of all runs in which a genome is involved
if args.show_genomes_runs:
statement = (
session.query(
Genome.genome_id,
Run.run_id,
Genome.description,
Genome.path,
Genome.genome_hash,
Label.label,
Label.class_label,
Run.name,
Run.method,
Run.date,
)
.join(rungenome, Run.run_id == rungenome.c.run_id)
.join(
Label,
and_(Genome.genome_id == Label.genome_id, Run.run_id == Label.run_id),
)
.order_by(Genome.genome_id, Run.run_id)
.statement
)
headers = [
"genome ID",
"run ID",
"genome description",
"genome path",
"genome hash",
"genome label",
"""Return dictionary of genome labels, keyed by row/column ID.
:param session: live SQLAlchemy session
:param run_id: the Run.run_id value for matrices
The labels should be valid for identity, coverage and other complete
matrix results accessed via the .df_* attributes of a run.
Labels are returned keyed by the string of the genome ID, for compatibility with
matplotlib.
"""
results = (
session.query(Genome.genome_id, Label.label)
.join(rungenome, Run)
.join(
Label, and_(Genome.genome_id == Label.genome_id, Run.run_id == Label.run_id)
)
.filter(Run.run_id == run_id)
.all()
)
return {str(_.genome_id): _.label for _ in results}