/
annotations.py
43 lines (38 loc) · 1.31 KB
/
annotations.py
1
2
3
4
5
6
7
8
9
10
11
12
13
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
from __future__ import annotations
from datetime import datetime
from typing import List, Optional
from sqlalchemy.orm import Query
from flexmeasures.data.models.annotations import (
Annotation,
GenericAssetAnnotationRelationship,
)
from flexmeasures.data.models.data_sources import DataSource
def query_asset_annotations(
asset_id: int,
annotations_after: Optional[datetime] = None,
annotations_before: Optional[datetime] = None,
sources: Optional[List[DataSource]] = None,
annotation_type: str | None = None,
) -> Query:
"""Match annotations assigned to the given asset."""
query = Annotation.query.join(GenericAssetAnnotationRelationship).filter(
GenericAssetAnnotationRelationship.generic_asset_id == asset_id,
GenericAssetAnnotationRelationship.annotation_id == Annotation.id,
)
if annotations_after is not None:
query = query.filter(
Annotation.end > annotations_after,
)
if annotations_before is not None:
query = query.filter(
Annotation.start < annotations_before,
)
if sources:
query = query.filter(
Annotation.source.in_(sources),
)
if annotation_type is not None:
query = query.filter(
Annotation.type == annotation_type,
)
return query