pre-review fixes
This commit is contained in:
@@ -268,8 +268,8 @@ class Repository:
|
||||
by_key.setdefault(r["key"], []).append(r["value"])
|
||||
return [DomainConstraint(key=k, allowed_values=v) for k, v in by_key.items()]
|
||||
|
||||
def get_domain(self, name: str) -> Domain | None:
|
||||
row = self.conn.execute("SELECT * FROM domains WHERE name = ?", (name,)).fetchone()
|
||||
def _load_domain(self, where: str, param: str | int) -> Domain | None:
|
||||
row = self.conn.execute(f"SELECT * FROM domains WHERE {where} = ?", (param,)).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
weights = self.conn.execute(
|
||||
@@ -296,37 +296,15 @@ class Repository:
|
||||
constraints=self._load_domain_constraints(row["id"]),
|
||||
)
|
||||
|
||||
def get_domain(self, name: str) -> Domain | None:
|
||||
return self._load_domain("name", name)
|
||||
|
||||
def list_domains(self) -> list[Domain]:
|
||||
rows = self.conn.execute("SELECT name FROM domains ORDER BY name").fetchall()
|
||||
return [self.get_domain(r["name"]) for r in rows]
|
||||
|
||||
def get_domain_by_id(self, domain_id: int) -> Domain | None:
|
||||
row = self.conn.execute("SELECT * FROM domains WHERE id = ?", (domain_id,)).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
weights = self.conn.execute(
|
||||
"""SELECT m.name, m.unit, dmw.weight, dmw.norm_min, dmw.norm_max,
|
||||
dmw.metric_id, dmw.lower_is_better
|
||||
FROM domain_metric_weights dmw
|
||||
JOIN metrics m ON dmw.metric_id = m.id
|
||||
WHERE dmw.domain_id = ?""",
|
||||
(row["id"],),
|
||||
).fetchall()
|
||||
return Domain(
|
||||
id=row["id"],
|
||||
name=row["name"],
|
||||
description=row["description"] or "",
|
||||
metric_bounds=[
|
||||
MetricBound(
|
||||
metric_name=w["name"], weight=w["weight"],
|
||||
norm_min=w["norm_min"], norm_max=w["norm_max"],
|
||||
metric_id=w["metric_id"], unit=w["unit"] or "",
|
||||
lower_is_better=bool(w["lower_is_better"]),
|
||||
)
|
||||
for w in weights
|
||||
],
|
||||
constraints=self._load_domain_constraints(row["id"]),
|
||||
)
|
||||
return self._load_domain("id", domain_id)
|
||||
|
||||
def update_domain(self, domain_id: int, name: str, description: str) -> None:
|
||||
self.conn.execute(
|
||||
|
||||
Reference in New Issue
Block a user