Handle failures

NotCompleted FALSE type

A FALSE type is returned when a condition is not met. For example, below we create an app that selects 2 specific sequences from an alignment. Applying this to a data set where a "Mouse" sequence does not exist produces a FALSE type.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import cogent3 as c3

aln = c3.get_dataset("primate-brca1")
select_seqs = c3.get_app("take_named_seqs", "Mouse", "Human")
result = select_seqs(aln)
assert result == False
print(result)

# NotCompleted(type=FAIL, origin=take_named_seqs, source="brca1", message="named
# seq(s) {'Mouse'} not in ('FlyingLem', 'TreeShrew', 'Galago', 'HowlerMon',
# 'Rhesus', 'Orangutan', 'Gorilla', 'Chimpanzee', 'Human')")

Inspecting NotCompleted attributes

The NotCompleted instance has attributes identifying what data failed:

result.source

Where the failure occurred:

result.origin

And the reason for the failure:

result.message

The .type attribute is the NotCompletedType enum value (e.g. NotCompletedType.FALSE, NotCompletedType.ERROR, or NotCompletedType.BUG).

NotCompleted ERROR type

An ERROR type is returned if an unexpected condition occurs, such as an exception raised during execution. Here we illustrate this by trying to open a file with an incorrect path.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import cogent3 as c3

reader = c3.get_app("load_aligned", moltype="dna")
result = reader("primate_brca1.fasta")
print(result)

# NotCompleted(type=ERROR, origin=load_aligned, source="primate_brca1",
# message="Traceback (most recent call last):   File
# "/home/runner/work/scinexus/scinexus/src/scinexus/composable.py", line 545, in
# __call__     result = self.main(val, *args, **kwargs)   File [...]

Composed functions propagate NotCompleted results

If you have a composed function with multiple steps and a failure occurs, the resulting NotCompleted is returned without any of the subsequent steps being applied. For example, we make a composed app from both of the above apps:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import cogent3 as c3

reader = c3.get_app("load_aligned", moltype="dna")
select_seqs = c3.get_app("take_named_seqs", "Mouse", "Human")
app = reader + select_seqs
result = app("data/primate_brca1.fasta")
print(result)

# NotCompleted(type=FAIL, origin=take_named_seqs, source="primate_brca1",
# message="named seq(s) {'Mouse'} not in ('FlyingLem', 'TreeShrew', 'Galago',
# 'HowlerMon', 'Rhesus', 'Orangutan', 'Gorilla', 'Chimpanzee', 'Human')")

The failure originated in select_seqs (an instance of take_named_seqs), and reader ran successfully — but the NotCompleted propagated through the rest of the pipeline.

1
2
3
4
5
6
7
8
9

import cogent3 as c3

reader = c3.get_app("load_aligned", moltype="dna")
select_seqs = c3.get_app("take_named_seqs", "Mouse", "Human")
app = reader + select_seqs
result = app("primate_brca1.fasta")
print(result)

# NotCompleted(type=ERROR, origin=load_aligned, source="primate_brca1", [...]

Here the failure originated in reader (bad path), and select_seqs was never called.

Creating NotCompleted in your own apps

You can return a NotCompleted from your own app to signal that a particular input cannot be processed:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from scinexus import define_app, NotCompleted, NotCompletedType


@define_app
def require_min_length(val: str, min_length: int = 10) -> str:
    if len(val) < min_length:
        return NotCompleted(
            NotCompletedType.FALSE,
            "require_min_length",
            val,
            message=f"too short: {len(val)} < {min_length}",
        )
    return val

Receiving NotCompleted with skip_not_completed=False

By default, apps skip NotCompleted inputs — they propagate without calling main(). If your app needs to see NotCompleted values (e.g. a writer that records failures), set skip_not_completed=False:

1
2
3
4
5
6
7
8
9

from scinexus import define_app, NotCompleted


@define_app(skip_not_completed=False)
def log_failures(val: str) -> str:
    if isinstance(val, NotCompleted):
        print(f"Failure: {val.message}")
        return val
    return val