/
parsing_utils.py
53 lines (47 loc) · 1.66 KB
/
parsing_utils.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
44
45
46
47
48
49
50
51
52
53
from __future__ import annotations
from collections.abc import Sequence
from flask import current_app
from flexmeasures.data import db
from flexmeasures.data.models.data_sources import DataSource
def parse_source_arg(
source: DataSource
| int
| str
| Sequence[DataSource]
| Sequence[int]
| Sequence[str]
| None,
) -> list[DataSource] | None:
"""Parse the "source" argument by looking up DataSources corresponding to any given ids or names.
Passes None as is (i.e. no source argument is given).
Accepts ids and names as list or tuples, always converting them to a list.
"""
if source is None:
return source
if isinstance(source, (DataSource, str, int)):
sources = [source]
else:
sources = source
parsed_sources: list[DataSource] = []
for source in sources:
if isinstance(source, int):
parsed_source = (
db.session.query(DataSource).filter_by(id=source).one_or_none()
)
if parsed_source is None:
current_app.logger.warning(
f"Beliefs searched for unknown source {source}"
)
else:
parsed_sources.append(parsed_source)
elif isinstance(source, str):
_parsed_sources = db.session.query(DataSource).filter_by(name=source).all()
if _parsed_sources is []:
current_app.logger.warning(
f"Beliefs searched for unknown source {source}"
)
else:
parsed_sources.extend(_parsed_sources)
else:
parsed_sources.append(source)
return parsed_sources