Skip to content

Application schemas

File for storing the application models

SingleImageModel

Bases: BaseModel

Schema for a single instance in the PredictRequest.

Exactly one of image (urlsafe base64) or url (http(s) location the API downloads itself) must be provided. Supplying both, or neither, is a validation error.

Source code in src/app/schemas.py 
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
class SingleImageModel(BaseModel):
    """
    Schema for a single instance in the PredictRequest.

    Exactly one of ``image`` (urlsafe base64) or ``url`` (http(s) location the API downloads
    itself) must be provided. Supplying both, or neither, is a validation error.
    """
    id: int = Field(description="Identifier of the image", examples=[1])
    image: Optional[str] = Field(default=None,
                                 description="Image as a urlsafe base64 string. Mutually "
                                             "exclusive with 'url'",
                                 examples=[b64_image])
    url: Optional[str] = Field(default=None,
                               description="Public http(s) url the API downloads the image from. "
                                           "Mutually exclusive with 'image'",
                               examples=["https://example.com/photo.jpg"])

    @model_validator(mode="after")
    def exactly_one_source(self):
        """Require exactly one of image / url (empty strings count as missing)."""
        has_image = bool(self.image)
        has_url = bool(self.url)
        if has_image == has_url:
            raise ValueError("Provide exactly one of 'image' (base64) or 'url' for each element")
        return self

exactly_one_source()

Require exactly one of image / url (empty strings count as missing).

Source code in src/app/schemas.py 
59
60
61
62
63
64
65
66
@model_validator(mode="after")
def exactly_one_source(self):
    """Require exactly one of image / url (empty strings count as missing)."""
    has_image = bool(self.image)
    has_url = bool(self.url)
    if has_image == has_url:
        raise ValueError("Provide exactly one of 'image' (base64) or 'url' for each element")
    return self

TrimImageResponse

Bases: BaseModel

Response of the trim_image endpoint. A separate model from :class:SingleImageModel so the output is not subject to the request-side image/url mutual-exclusion validation.

Source code in src/app/schemas.py 
69
70
71
72
73
74
75
76
class TrimImageResponse(BaseModel):
    """
    Response of the trim_image endpoint. A separate model from :class:`SingleImageModel` so the
    output is not subject to the request-side image/url mutual-exclusion validation.
    """
    id: int = Field(description="Identifier of the image", examples=[1])
    image: str = Field(description="Trimmed image as a urlsafe base64 PNG string",
                       examples=[b64_image])

TrimImageRequest

Bases: SingleImageModel

Schema for trim_image endpoint. Accepts exactly one of image (base64) or url (inherited from :class:SingleImageModel); trimming operates on the full-resolution original.

Source code in src/app/schemas.py 
79
80
81
82
83
84
85
86
87
88
89
class TrimImageRequest(SingleImageModel):
    """
    Schema for trim_image endpoint. Accepts exactly one of ``image`` (base64) or ``url`` (inherited
    from :class:`SingleImageModel`); trimming operates on the full-resolution original.
    """
    norm_thresh: Optional[confloat(ge=0.0)] = Field(default=1.5,
                                                    description="Threshold of vectors norm to "
                                                                "consider vectors as a same")
    initial_thresh: Optional[confloat(ge=0.0)] = Field(default=0.001,
                                                       description="Threshold of vectors norm "
                                                                   "to identify trimming direction")

PredictBatchRequest

Bases: BaseModel

Batch of images with id and image field each

Source code in src/app/schemas.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class PredictBatchRequest(BaseModel):
    """
    Batch of images with id and image field each
    """
    images: conlist(item_type=SingleImageModel, min_length=1) = \
        Field(description="List of images to infer the CNN on")
    core_listing_id: Optional[int] = Field(default=None, description="Identifier of the house the images belong to",
                                           examples=[1])
    thresholds: Optional[Union[confloat(ge=0, le=1), Dict[str, confloat(ge=0, le=1)]]] = Field(default=None,
                                                                                               description="Probability threshold to use when extracting tags via multilabel network. "
                                                                                                           "If not given, defaults from the config file will be used",
                                                                                               examples=[{
                                                                                                   "interior": 0.5}])

    @field_validator("images")
    def ids_are_unique(cls, v):
        """Check if all ids are unique in a list of elements"""
        ids = [item.id for item in v]
        validate_unique_ids(ids=ids, field_name='id')
        return v

ids_are_unique(v)

Check if all ids are unique in a list of elements

Source code in src/app/schemas.py 
106
107
108
109
110
111
@field_validator("images")
def ids_are_unique(cls, v):
    """Check if all ids are unique in a list of elements"""
    ids = [item.id for item in v]
    validate_unique_ids(ids=ids, field_name='id')
    return v

PredictHomeRequest

Bases: PredictBatchRequest

Extended model for the endpoint that processes all house images

Source code in src/app/schemas.py 
114
115
116
117
118
119
120
121
class PredictHomeRequest(PredictBatchRequest):
    """Extended model for the endpoint that processes all house images"""
    publish_embeddings: bool = Field(default=False,
                                     description="Whether to send vector embeddings to the message broker queue")
    embeddings_compression: CompressionMethod = Field(default=CompressionMethod.GZIP,
                                                      description="Compression method to use when sending "
                                                                  "embeddings to the message broker queue",
                                                      examples=[CompressionMethod.GZIP])

Prediction

Bases: BaseModel

Model for prediction of a single image

Source code in src/app/schemas.py 
124
125
126
127
128
129
130
131
132
133
134
135
class Prediction(BaseModel):
    """
    Model for prediction of a single image
    """
    scores: Union[List[float], Dict[str, float], List[dict]] = \
        Field(description="Dictionary with present labels and their probabilities",
              examples=[[{"name": "class1", "probability": 0.75},
                         {"name": "class4", "probability": 0.6}]]
              )
    message: Optional[str] = Field(default=None, description="Error message (if any) from the final network",
                                   examples=[''])
    status: bool = Field(description="Boolean showing whether the image is successfully processed", examples=[True])

TaggedImageModel

Bases: BaseModel

Model for a single tagged image

Source code in src/app/schemas.py 
138
139
140
141
142
143
144
145
146
147
148
class TaggedImageModel(BaseModel):
    """
    Model for a single tagged image
    """
    id: int = Field(description="Identifier of the image", examples=[1])
    results: Dict[str, Prediction] = Field(description="Results per network",
                                           examples=[{"net1": [{"name": "class1", "probability": 0.55}],
                                                      "net2": [0.25,
                                                               0.75,
                                                               0.1,
                                                               0.4]}])

PredictBatchResponse

Bases: BaseModel

Response for prediction request on a bunch of (unrelated) images

Source code in src/app/schemas.py 
151
152
153
154
155
156
class PredictBatchResponse(BaseModel):
    """
    Response for prediction request on a bunch of (unrelated) images
    """
    predictions: List[TaggedImageModel] = Field(description="List of prediction dicts for each "
                                                            "passed image")

HouseFeatures

Bases: BaseModel

Model for extracted home features

Source code in src/app/schemas.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class HouseFeatures(BaseModel):
    """
    Model for extracted home features
    """

    exterior_styles: List[dict] = Field(description="Dictionary of Exterior Styles as keys and their averaged "
                                                    "probability over all facade images as values",
                                        examples=[[{"name": "style1", "probability": 0.1},
                                                   {"name": "style2", "probability": 0.25},
                                                   {"name": "style3", "probability": 0.65}]])
    prob_ranking: List[dict] = Field(description="Dictionary of present classes and the array of ids of images "
                                                 "containing them, sorted in descending order by probability",
                                     examples=[[{"name": "feature1", "ordering": [0, 1, 2, 3]},
                                                {"name": "feature5", "ordering": [2, 4, 3]}]]
                                     )

PredictHomeResponse

Bases: BaseModel

A model for house info prediction endpoint response

Source code in src/app/schemas.py 
176
177
178
179
180
181
182
183
184
185
186
class PredictHomeResponse(BaseModel):
    """
    A model for house info prediction endpoint response
    """
    core_listing_id: Optional[int] = Field(default=None, description="Identifier of a house", examples=[1])
    status: bool = Field(description="Whether at least one image is processed successfully", examples=[True])
    house_info: HouseFeatures = Field(description="House-specific information extracted from the "
                                                  "passed images")
    results_per_image: List[TaggedImageModel] = Field(description="List of predictions for "
                                                                  "successfully processed images")
    publish_results: PublishResults | None = Field(None, description="Information about the published message if any")

Dependence

Bases: BaseModel

Dependencies model

Source code in src/app/schemas.py 
189
190
191
192
193
194
195
196
197
198
199
200
class Dependence(BaseModel):
    """
    Dependencies model
    """

    name: str = Field(description="Name of the dependency", examples=["some_dependency"])
    version: Optional[int] = Field(default=None, description="Version of the dependency", examples=[1])
    connected: bool = Field(description="Boolean representing whether API is succesfully "
                                        "connected to said dependency", examples=[True])
    status: Optional[str] = Field(default=None, description="Current status of the dependence returned by the "
                                                            "final endpoint", examples=[True])
    message: Optional[str] = Field(default="success", description="Message based on current status", examples=[''])

HealthCheckResponse

Bases: BaseModel

HealthCheck endpoint response model

Source code in src/app/schemas.py 
203
204
205
206
207
208
209
class HealthCheckResponse(BaseModel):
    """
    HealthCheck endpoint response model
    """

    healthy: bool = Field(description="Boolean showing whether the service is healthy", examples=[True])
    dependencies: List[Dependence] = Field(description="List of dicts for each dependency")

InfoResponse

Bases: BaseModel

Response for info endpoint returning name/url/output info on each network

Source code in src/app/schemas.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class InfoResponse(BaseModel):
    """
    Response for info endpoint returning name/url/output info on each network
    """
    networks: List[dict]
    model_config = ConfigDict(json_schema_extra={
        'example': {
            "networks": [
                {"name": "network1",
                 "network": {
                     "predict_url": "localhost:8501/v1/models/network1/versions/2:predict",
                     "version": 2, "num_outputs": 101,
                     "raw_output_names": ["class1", "class2", "class3"],
                     "is_multilabel": True},
                 "outputs": {"num_classes": 3,
                             "all_classes": ["class1", "class2", "class3"],
                             "default_thresholds": {"class1": 0.5,
                                                    "class2": 0.65,
                                                    "class3": 0.7}}},
                {"name": "network2",
                 "network": {
                     "predict_url": "localhost:8501/v1/models/network2/versions/2:predict",
                     "version": 2, "num_outputs": 2,
                     "raw_output_names": ["class1", "class2"],
                     "outputs": {"num_classes": 2,
                                 "all_classes": ["class1", "class2"]}}},
                {"name": "network3", "network": {
                    "predict_url": "localhost:8501/v1/models/network3/versions/1:predict",
                    "version": 1}}]}})

validate_unique_ids(ids, field_name='id')

Validate whether provided elements are unique

Parameters:

Name Type Description Default
ids Collection

Elements to validate

 required
field_name str = 'id'

Name of the field. Used in the error message

 'id'

Returns:

Type Description
None

Raises:

Type Description
ValueError : if elements are non-unique
Source code in src/app/schemas.py 
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
def validate_unique_ids(ids: Collection, field_name: str = 'id') -> None:
    """
    Validate whether provided elements are unique

    Parameters
    ----------
    ids : Collection
        Elements to validate
    field_name : str = 'id'
        Name of the field. Used in the error message

    Returns
    -------
    None

    Raises
    ------
    ValueError : if elements are non-unique
    """
    if not len(set(ids)) == len(ids):
        # Validation failed, build an error message
        counts = Counter(ids)
        non_unique = [id_ for id_, count in counts.items() if count > 1]

        raise ValueError(f"Please provide unique {field_name} for each element. Non unique ids: {non_unique}")