Skip to content

NeuroMorpho Queries

neuromorphopy.query

Query

Build and validate NeuroMorpho queries.

Source code in neuromorphopy/query.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class Query:
    """Build and validate NeuroMorpho queries."""

    def __init__(self) -> None:
        self._config = QueryConfig()

    @classmethod
    def from_file(cls, path: Path | str) -> dict[str, list[str]]:
        """Create and build query directly from JSON or YAML file."""
        path = Path(path)

        if not path.exists():
            raise FileNotFoundError(f"Config file not found: {path}")

        if path.suffix.lower() in {".yml", ".yaml"}:
            data = yaml.safe_load(path.read_text())
        elif path.suffix.lower() == ".json":
            data = json.loads(path.read_text())
        else:
            raise ValueError("Config file must be JSON or YAML")

        config = QueryConfig.model_validate(data)
        builder = cls()
        builder._config = config
        return builder.build()

    def filter(self, field: str, values: str | list[str]) -> "Query":
        """Add a filter with validation."""
        if isinstance(values, str):
            values = [values]

        self._config.validate_field(field, values)
        filters = dict(self._config.filters)
        filters[field] = values
        self._config = QueryConfig(filters=filters, sort=self._config.sort)
        return self

    def sort(self, field: str, ascending: bool = True) -> "Query":
        """Add sorting with validation."""
        if field not in QueryFields.get_fields():
            raise ValueError(f"Invalid field: {field}")

        self._config = QueryConfig(
            filters=self._config.filters, sort=QuerySort(field=field, ascending=ascending)
        )
        return self

    def build(self) -> dict[str, Any]:
        """Build the final query dictionary."""
        query: dict[str, Any] = dict(self._config.filters)
        if self._config.sort:
            query["_sort"] = {"field": self._config.sort.field, "order": self._config.sort.order}

        return query

build()

Build the final query dictionary.

Source code in neuromorphopy/query.py 
61
62
63
64
65
66
67
def build(self) -> dict[str, Any]:
    """Build the final query dictionary."""
    query: dict[str, Any] = dict(self._config.filters)
    if self._config.sort:
        query["_sort"] = {"field": self._config.sort.field, "order": self._config.sort.order}

    return query

filter(field, values)

Add a filter with validation.

Source code in neuromorphopy/query.py 
40
41
42
43
44
45
46
47
48
49
def filter(self, field: str, values: str | list[str]) -> "Query":
    """Add a filter with validation."""
    if isinstance(values, str):
        values = [values]

    self._config.validate_field(field, values)
    filters = dict(self._config.filters)
    filters[field] = values
    self._config = QueryConfig(filters=filters, sort=self._config.sort)
    return self

from_file(path) classmethod

Create and build query directly from JSON or YAML file.

Source code in neuromorphopy/query.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@classmethod
def from_file(cls, path: Path | str) -> dict[str, list[str]]:
    """Create and build query directly from JSON or YAML file."""
    path = Path(path)

    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {path}")

    if path.suffix.lower() in {".yml", ".yaml"}:
        data = yaml.safe_load(path.read_text())
    elif path.suffix.lower() == ".json":
        data = json.loads(path.read_text())
    else:
        raise ValueError("Config file must be JSON or YAML")

    config = QueryConfig.model_validate(data)
    builder = cls()
    builder._config = config
    return builder.build()

sort(field, ascending=True)

Add sorting with validation.

Source code in neuromorphopy/query.py 
51
52
53
54
55
56
57
58
59
def sort(self, field: str, ascending: bool = True) -> "Query":
    """Add sorting with validation."""
    if field not in QueryFields.get_fields():
        raise ValueError(f"Invalid field: {field}")

    self._config = QueryConfig(
        filters=self._config.filters, sort=QuerySort(field=field, ascending=ascending)
    )
    return self

QueryConfig

Bases: BaseModel

Query configuration with validation.

Source code in neuromorphopy/query.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class QueryConfig(BaseModel):
    """Query configuration with validation."""

    filters: dict[str, list[str]] = Field(default_factory=dict)
    sort: QuerySort | None = None

    @classmethod
    @lru_cache(maxsize=1)
    def get_valid_fields(cls) -> set[str]:
        """Get all valid fields from NeuroMorpho API."""
        response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields")
        return set(response.json()["Neuron Fields"])

    @classmethod
    @lru_cache(maxsize=100)
    def get_field_values(cls, field: str) -> set[str]:
        """Get valid values for a specific field."""
        response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields/{field}")
        return set(response.json()["fields"])

    @staticmethod
    def validate_field(field: str, values: list[str]) -> None:
        """Validate a single field and its values."""
        if field not in QueryFields.get_fields():
            raise ValueError(f"Invalid field: {field}")

        valid_values = QueryFields.get_values(field)
        invalid_values = set(values) - valid_values
        if invalid_values:
            raise ValueError(
                f"Invalid values for {field}: {invalid_values}\nValid values are: {valid_values}"
            )

get_field_values(field) cached classmethod

Get valid values for a specific field.

Source code in neuromorphopy/query.py 
128
129
130
131
132
133
@classmethod
@lru_cache(maxsize=100)
def get_field_values(cls, field: str) -> set[str]:
    """Get valid values for a specific field."""
    response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields/{field}")
    return set(response.json()["fields"])

get_valid_fields() cached classmethod

Get all valid fields from NeuroMorpho API.

Source code in neuromorphopy/query.py 
121
122
123
124
125
126
@classmethod
@lru_cache(maxsize=1)
def get_valid_fields(cls) -> set[str]:
    """Get all valid fields from NeuroMorpho API."""
    response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields")
    return set(response.json()["Neuron Fields"])

validate_field(field, values) staticmethod

Validate a single field and its values.

Source code in neuromorphopy/query.py 
135
136
137
138
139
140
141
142
143
144
145
146
@staticmethod
def validate_field(field: str, values: list[str]) -> None:
    """Validate a single field and its values."""
    if field not in QueryFields.get_fields():
        raise ValueError(f"Invalid field: {field}")

    valid_values = QueryFields.get_values(field)
    invalid_values = set(values) - valid_values
    if invalid_values:
        raise ValueError(
            f"Invalid values for {field}: {invalid_values}\nValid values are: {valid_values}"
        )

QueryFields

Helper class for accessing NeuroMorpho query field information.

Source code in neuromorphopy/query.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class QueryFields:
    """Helper class for accessing NeuroMorpho query field information."""

    @classmethod
    @lru_cache(maxsize=1)
    def get_fields(cls) -> set[str]:
        """Get all valid query fields from NeuroMorpho API."""
        response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields")
        return set(response.json()["Neuron Fields"])

    @classmethod
    @lru_cache(maxsize=100)
    def get_values(cls, field: str) -> set[str]:
        """Get valid values for a specific field."""
        if field not in cls.get_fields():
            raise ValidationError(f"Invalid field: {field}")

        try:
            response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields/{field}")
            return set(response.json()["fields"])
        except Exception as err:
            raise ApiError(f"Failed to fetch values for field {field}: {err!s}") from err

    @classmethod
    def describe(cls) -> dict[str, set[str]]:
        """Get all fields and their valid values."""
        return {field: cls.get_values(field) for field in cls.get_fields()}

describe() classmethod

Get all fields and their valid values.

Source code in neuromorphopy/query.py 
93
94
95
96
@classmethod
def describe(cls) -> dict[str, set[str]]:
    """Get all fields and their valid values."""
    return {field: cls.get_values(field) for field in cls.get_fields()}

get_fields() cached classmethod

Get all valid query fields from NeuroMorpho API.

Source code in neuromorphopy/query.py 
73
74
75
76
77
78
@classmethod
@lru_cache(maxsize=1)
def get_fields(cls) -> set[str]:
    """Get all valid query fields from NeuroMorpho API."""
    response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields")
    return set(response.json()["Neuron Fields"])

get_values(field) cached classmethod

Get valid values for a specific field.

Source code in neuromorphopy/query.py 
80
81
82
83
84
85
86
87
88
89
90
91
@classmethod
@lru_cache(maxsize=100)
def get_values(cls, field: str) -> set[str]:
    """Get valid values for a specific field."""
    if field not in cls.get_fields():
        raise ValidationError(f"Invalid field: {field}")

    try:
        response = request_url_get(f"{NEUROMORPHO_API}/neuron/fields/{field}")
        return set(response.json()["fields"])
    except Exception as err:
        raise ApiError(f"Failed to fetch values for field {field}: {err!s}") from err