Skip to content

gateways

Pydantic models/schemas for the Gateways resource.

GatewayCreate

Bases: EntryResourceCreate, GatewayResourceAttributes

Model for creating new Gateway resources in the MongoDB

Source code in optimade_gateway/models/gateways.py 
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
147
148
149
150
151
152
153
154
155
class GatewayCreate(EntryResourceCreate, GatewayResourceAttributes):
    """Model for creating new Gateway resources in the MongoDB"""

    id: Annotated[
        str | None,
        OptimadeField(
            description=EntryResource.model_fields["id"].description,
            support=EntryResource.model_fields["id"].json_schema_extra[
                "x-optimade-support"
            ],
            queryable=EntryResource.model_fields["id"].json_schema_extra[
                "x-optimade-queryable"
            ],
            pattern=r"^[^/]*$",  # This pattern is the special addition
        ),
    ] = None

    database_ids: Annotated[
        set[str] | None,
        Field(description="A unique list of database IDs for registered databases."),
    ] = None

    databases: Annotated[
        list[LinksResource] | None,
        Field(
            description=GatewayResourceAttributes.model_fields["databases"].description
        ),
    ] = None  # type: ignore[assignment]

    @model_validator(mode="after")
    def specify_databases(self) -> GatewayCreate:
        """Either `database_ids` or `databases` must be non-empty.
        Both together is also fine.
        """
        if not any(getattr(self, field) for field in ("database_ids", "databases")):
            raise ValueError("Either 'database_ids' or 'databases' MUST be specified")
        return self

database_ids: Annotated[set[str] | None, Field(description='A unique list of database IDs for registered databases.')] = None class-attribute instance-attribute

databases: Annotated[list[LinksResource] | None, Field(description=GatewayResourceAttributes.model_fields['databases'].description)] = None class-attribute instance-attribute

id: Annotated[str | None, OptimadeField(description=EntryResource.model_fields['id'].description, support=EntryResource.model_fields['id'].json_schema_extra['x-optimade-support'], queryable=EntryResource.model_fields['id'].json_schema_extra['x-optimade-queryable'], pattern='^[^/]*$')] = None class-attribute instance-attribute

specify_databases()

Either database_ids or databases must be non-empty. Both together is also fine.

Source code in optimade_gateway/models/gateways.py 
148
149
150
151
152
153
154
155
@model_validator(mode="after")
def specify_databases(self) -> GatewayCreate:
    """Either `database_ids` or `databases` must be non-empty.
    Both together is also fine.
    """
    if not any(getattr(self, field) for field in ("database_ids", "databases")):
        raise ValueError("Either 'database_ids' or 'databases' MUST be specified")
    return self

GatewayResource

Bases: EntryResource

OPTIMADE gateway

A resource representing a dynamic collection of OPTIMADE databases. The gateway can be treated as any other OPTIMADE gateway, but the entries are an aggregate of multiple databases. The id of each aggregated resource will reflect the originating database.

Source code in optimade_gateway/models/gateways.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class GatewayResource(EntryResource):
    """OPTIMADE gateway

    A resource representing a dynamic collection of OPTIMADE databases.
    The gateway can be treated as any other OPTIMADE gateway, but the entries are an
    aggregate of multiple databases. The `id` of each aggregated resource will reflect
    the originating database.
    """

    id: Annotated[
        str,
        OptimadeField(
            description=EntryResource.model_fields["id"].description,
            support=EntryResource.model_fields["id"].json_schema_extra[
                "x-optimade-support"
            ],
            queryable=EntryResource.model_fields["id"].json_schema_extra[
                "x-optimade-queryable"
            ],
            pattern=r"^[^/]*$",
        ),
    ]

    type: Annotated[
        Literal["gateways"],
        Field(description="The name of the type of an entry."),
    ] = "gateways"

    attributes: Annotated[
        GatewayResourceAttributes,
        Field(description=EntryResource.model_fields["attributes"].description),
    ]

attributes: Annotated[GatewayResourceAttributes, Field(description=EntryResource.model_fields['attributes'].description)] instance-attribute

id: Annotated[str, OptimadeField(description=EntryResource.model_fields['id'].description, support=EntryResource.model_fields['id'].json_schema_extra['x-optimade-support'], queryable=EntryResource.model_fields['id'].json_schema_extra['x-optimade-queryable'], pattern='^[^/]*$')] instance-attribute

type: Annotated[Literal['gateways'], Field(description='The name of the type of an entry.')] = 'gateways' class-attribute instance-attribute

GatewayResourceAttributes

Bases: EntryResourceAttributes

Attributes for an OPTIMADE gateway

Source code in optimade_gateway/models/gateways.py 
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class GatewayResourceAttributes(EntryResourceAttributes):
    """Attributes for an OPTIMADE gateway"""

    databases: Annotated[
        list[LinksResource],
        Field(
            description=(
                "List of databases (OPTIMADE 'links') to be queried in this gateway."
            ),
        ),
    ]

    @field_validator("databases", mode="after")
    @classmethod
    def unique_base_urls(cls, value: list[LinksResource]) -> list[LinksResource]:
        """Remove extra entries with repeated base_urls.

        Also, ensure databases are not of type `"root"` or `"providers"`

        !!! note
            Both `"external"` and `"child"` can still represent index meta-dbs,
            but `"root"` and `"providers"` can not represent "regular" dbs.

        """
        for resource in value:
            if resource.attributes.link_type in (LinkType.ROOT, LinkType.PROVIDERS):
                raise ValueError(
                    "Databases with 'root' or 'providers' link_type is not allowed for "
                    f"gateway resources. Given database: {resource}"
                )

        db_base_urls = [_.attributes.base_url for _ in value]

        unique_base_urls = []
        for base_url in db_base_urls:
            if base_url is None:
                continue
            if base_url in unique_base_urls:
                continue
            unique_base_urls.append(base_url)

        if len(db_base_urls) == len(unique_base_urls):
            return value

        repeated_base_urls = [_ for _ in unique_base_urls if db_base_urls.count(_) > 1]
        new_databases = [
            _ for _ in value if _.attributes.base_url not in repeated_base_urls
        ]
        for base_url in repeated_base_urls:
            new_databases.append(
                next(_ for _ in value if _.attributes.base_url == base_url)
            )
        warnings.warn(
            "Removed extra database entries for a gateway, because the base_url was "
            "repeated. The first found database entry was kept, while the others were "
            f"removed. Original number of databases: {len(value)}. New number of "
            f"databases: {len(new_databases)} Repeated base_urls (number of repeats): "
            "{}".format(
                [
                    f"{base_url} ({db_base_urls.count(base_url)})"
                    for base_url in repeated_base_urls
                ]
            ),
            OptimadeGatewayWarning,
        )
        return new_databases

databases: Annotated[list[LinksResource], Field(description="List of databases (OPTIMADE 'links') to be queried in this gateway.")] instance-attribute

unique_base_urls(value) classmethod

Remove extra entries with repeated base_urls.

Also, ensure databases are not of type "root" or "providers"

Note

Both "external" and "child" can still represent index meta-dbs, but "root" and "providers" can not represent "regular" dbs.

Source code in optimade_gateway/models/gateways.py 
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@field_validator("databases", mode="after")
@classmethod
def unique_base_urls(cls, value: list[LinksResource]) -> list[LinksResource]:
    """Remove extra entries with repeated base_urls.

    Also, ensure databases are not of type `"root"` or `"providers"`

    !!! note
        Both `"external"` and `"child"` can still represent index meta-dbs,
        but `"root"` and `"providers"` can not represent "regular" dbs.

    """
    for resource in value:
        if resource.attributes.link_type in (LinkType.ROOT, LinkType.PROVIDERS):
            raise ValueError(
                "Databases with 'root' or 'providers' link_type is not allowed for "
                f"gateway resources. Given database: {resource}"
            )

    db_base_urls = [_.attributes.base_url for _ in value]

    unique_base_urls = []
    for base_url in db_base_urls:
        if base_url is None:
            continue
        if base_url in unique_base_urls:
            continue
        unique_base_urls.append(base_url)

    if len(db_base_urls) == len(unique_base_urls):
        return value

    repeated_base_urls = [_ for _ in unique_base_urls if db_base_urls.count(_) > 1]
    new_databases = [
        _ for _ in value if _.attributes.base_url not in repeated_base_urls
    ]
    for base_url in repeated_base_urls:
        new_databases.append(
            next(_ for _ in value if _.attributes.base_url == base_url)
        )
    warnings.warn(
        "Removed extra database entries for a gateway, because the base_url was "
        "repeated. The first found database entry was kept, while the others were "
        f"removed. Original number of databases: {len(value)}. New number of "
        f"databases: {len(new_databases)} Repeated base_urls (number of repeats): "
        "{}".format(
            [
                f"{base_url} ({db_base_urls.count(base_url)})"
                for base_url in repeated_base_urls
            ]
        ),
        OptimadeGatewayWarning,
    )
    return new_databases