715 lines
25 KiB
Python
715 lines
25 KiB
Python
"""PostgreSQL persistence for the inventory domain."""
|
|
|
|
from collections.abc import Callable, Iterator, Mapping, Sequence
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
|
|
from app.config import settings
|
|
|
|
|
|
ConnectionFactory = Callable[[], psycopg.Connection]
|
|
|
|
TAG_TRIMMED_NAME_SQL = (
|
|
"regexp_replace(name, '^[[:space:]]+|[[:space:]]+$', '', 'g')"
|
|
)
|
|
|
|
MINER_SELECT = f"""
|
|
SELECT
|
|
miners.id,
|
|
miners.site_id,
|
|
miners.room_id,
|
|
miners.asset_tag,
|
|
miners.current_mac,
|
|
host(miners.current_ip) AS current_ip,
|
|
miners.hostname,
|
|
miners.miner_sn,
|
|
miners.control_board_sn,
|
|
miners.power_sn,
|
|
miners.model,
|
|
miners.firmware_version,
|
|
miners.lifecycle_status,
|
|
miners.status,
|
|
miners.notes,
|
|
miners.first_seen_at,
|
|
miners.last_seen_at,
|
|
miners.created_at,
|
|
miners.updated_at,
|
|
tag_data.tags
|
|
FROM miners
|
|
LEFT JOIN LATERAL (
|
|
SELECT COALESCE(
|
|
jsonb_agg(
|
|
jsonb_build_object(
|
|
'id', tags.id,
|
|
'name', tags.name,
|
|
'created_at', tags.created_at,
|
|
'updated_at', tags.updated_at
|
|
)
|
|
ORDER BY lower({TAG_TRIMMED_NAME_SQL}), tags.id
|
|
),
|
|
'[]'::jsonb
|
|
) AS tags
|
|
FROM miner_tags
|
|
JOIN tags ON tags.id = miner_tags.tag_id
|
|
WHERE miner_tags.miner_id = miners.id
|
|
) tag_data ON true
|
|
"""
|
|
|
|
MINER_CURRENT_SELECT = f"""
|
|
SELECT
|
|
miners.id,
|
|
miners.site_id,
|
|
miners.room_id,
|
|
miners.asset_tag,
|
|
miners.current_mac,
|
|
host(miners.current_ip) AS current_ip,
|
|
miners.hostname,
|
|
miners.miner_sn,
|
|
miners.control_board_sn,
|
|
miners.power_sn,
|
|
miners.model,
|
|
miners.firmware_version,
|
|
miners.lifecycle_status,
|
|
miners.status,
|
|
miners.notes,
|
|
miners.first_seen_at,
|
|
miners.last_seen_at,
|
|
miners.created_at,
|
|
miners.updated_at,
|
|
tag_data.tags,
|
|
latest.observed_at,
|
|
latest.hashrate_ths,
|
|
latest.power_w,
|
|
latest.chip_temp_max_c,
|
|
latest.hash_stable
|
|
FROM miners
|
|
LEFT JOIN LATERAL (
|
|
SELECT COALESCE(
|
|
jsonb_agg(
|
|
jsonb_build_object(
|
|
'id', tags.id,
|
|
'name', tags.name,
|
|
'created_at', tags.created_at,
|
|
'updated_at', tags.updated_at
|
|
)
|
|
ORDER BY lower({TAG_TRIMMED_NAME_SQL}), tags.id
|
|
),
|
|
'[]'::jsonb
|
|
) AS tags
|
|
FROM miner_tags
|
|
JOIN tags ON tags.id = miner_tags.tag_id
|
|
WHERE miner_tags.miner_id = miners.id
|
|
) tag_data ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT snapshots.observed_at,
|
|
ROUND((snapshots.mhs_av / 1000000.0)::numeric, 2)::float
|
|
AS hashrate_ths,
|
|
ROUND(snapshots.power_w::numeric, 2)::float AS power_w,
|
|
ROUND(snapshots.chip_temp_max_c::numeric, 2)::float
|
|
AS chip_temp_max_c,
|
|
snapshots.hash_stable
|
|
FROM miner_status_snapshots AS snapshots
|
|
WHERE snapshots.miner_id = miners.id
|
|
ORDER BY snapshots.observed_at DESC, snapshots.id DESC
|
|
LIMIT 1
|
|
) latest ON true
|
|
"""
|
|
|
|
ACTIVE_REPAIR_STATUSES = (
|
|
"OPEN",
|
|
"IN_PROGRESS",
|
|
"WAITING_FOR_PARTS",
|
|
"READY_FOR_TEST",
|
|
)
|
|
|
|
OPERATIONAL_STATE_SQL = """
|
|
CASE
|
|
WHEN active_repair.id IS NOT NULL OR miners.lifecycle_status = 'in_repair'
|
|
THEN 'repair'
|
|
WHEN miners.lifecycle_status = 'retired' THEN 'retired'
|
|
WHEN miners.lifecycle_status = 'spare' THEN 'spare'
|
|
WHEN lower(trim(miners.status)) = 'offline' THEN 'offline'
|
|
WHEN latest.hash_stable = false THEN 'unstable'
|
|
WHEN lower(trim(miners.status)) = 'online' THEN 'online'
|
|
ELSE 'unknown'
|
|
END
|
|
"""
|
|
|
|
FLEET_SELECT = f"""
|
|
SELECT
|
|
miners.id, miners.asset_tag, miners.current_mac,
|
|
host(miners.current_ip) AS current_ip, miners.model, miners.miner_sn,
|
|
miners.lifecycle_status, miners.status, miners.created_at,
|
|
miners.updated_at, miners.last_seen_at, tag_data.tags,
|
|
latest.observed_at, latest.hashrate_ths, latest.power_w,
|
|
latest.chip_temp_max_c, latest.hash_stable,
|
|
active_repair.id AS active_repair_id,
|
|
active_repair.status AS active_repair_status,
|
|
active_repair.reason AS active_repair_reason,
|
|
{{location_fields}}
|
|
FROM miners
|
|
LEFT JOIN LATERAL (
|
|
SELECT COALESCE(
|
|
jsonb_agg(
|
|
jsonb_build_object(
|
|
'id', tags.id, 'name', tags.name,
|
|
'created_at', tags.created_at, 'updated_at', tags.updated_at
|
|
) ORDER BY lower({TAG_TRIMMED_NAME_SQL}), tags.id
|
|
), '[]'::jsonb
|
|
) AS tags
|
|
FROM miner_tags JOIN tags ON tags.id = miner_tags.tag_id
|
|
WHERE miner_tags.miner_id = miners.id
|
|
) tag_data ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT snapshots.observed_at,
|
|
ROUND((snapshots.mhs_av / 1000000.0)::numeric, 2)::float AS hashrate_ths,
|
|
ROUND(snapshots.power_w::numeric, 2)::float AS power_w,
|
|
ROUND(snapshots.chip_temp_max_c::numeric, 2)::float AS chip_temp_max_c,
|
|
snapshots.hash_stable
|
|
FROM miner_status_snapshots snapshots
|
|
WHERE snapshots.miner_id = miners.id
|
|
ORDER BY snapshots.observed_at DESC, snapshots.id DESC
|
|
LIMIT 1
|
|
) latest ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT repairs.id, repairs.status, repairs.reason
|
|
FROM repairs
|
|
WHERE repairs.miner_id = miners.id AND repairs.status = ANY(%s)
|
|
ORDER BY repairs.opened_at, repairs.id
|
|
LIMIT 1
|
|
) active_repair ON true
|
|
{{location_join}}
|
|
"""
|
|
|
|
LOCATION_FIELDS = """
|
|
mapped.site_id, mapped.site_name, mapped.room_id, mapped.room_name,
|
|
mapped.location_position_id, mapped.rack_label, mapped.shelf_label,
|
|
mapped.position_label, mapped.position_index
|
|
"""
|
|
|
|
REDACTED_LOCATION_FIELDS = """
|
|
NULL::uuid AS site_id, NULL::text AS site_name,
|
|
NULL::uuid AS room_id, NULL::text AS room_name,
|
|
NULL::uuid AS location_position_id, NULL::text AS rack_label,
|
|
NULL::text AS shelf_label, NULL::text AS position_label,
|
|
NULL::integer AS position_index
|
|
"""
|
|
|
|
LOCATION_JOIN = """
|
|
LEFT JOIN LATERAL (
|
|
SELECT sites.id AS site_id, sites.name AS site_name, rooms.id AS room_id,
|
|
COALESCE(rooms.name, lp.room_name) AS room_name,
|
|
lp.id AS location_position_id, lp.rack_name AS rack_label,
|
|
lp.shelf_name AS shelf_label, lp.position_label, lp.position_index
|
|
FROM location_positions lp
|
|
LEFT JOIN rooms ON rooms.id = lp.room_id
|
|
LEFT JOIN sites ON sites.id = rooms.site_id
|
|
WHERE lp.miner_id = miners.id
|
|
ORDER BY lp.id
|
|
LIMIT 1
|
|
) mapped ON true
|
|
"""
|
|
|
|
MINER_WRITE_COLUMNS = {
|
|
"site_id",
|
|
"room_id",
|
|
"asset_tag",
|
|
"current_mac",
|
|
"current_ip",
|
|
"hostname",
|
|
"miner_sn",
|
|
"control_board_sn",
|
|
"power_sn",
|
|
"model",
|
|
"firmware_version",
|
|
"lifecycle_status",
|
|
"status",
|
|
"notes",
|
|
"first_seen_at",
|
|
"last_seen_at",
|
|
}
|
|
|
|
|
|
def default_connection_factory() -> psycopg.Connection:
|
|
return psycopg.connect(
|
|
host=settings.postgres_host,
|
|
port=settings.postgres_port,
|
|
dbname=settings.postgres_db,
|
|
user=settings.postgres_user,
|
|
password=settings.postgres_password,
|
|
connect_timeout=5,
|
|
)
|
|
|
|
|
|
class InventoryRepository:
|
|
"""Database-only inventory persistence with optional transaction binding."""
|
|
|
|
def __init__(
|
|
self,
|
|
connection_factory: ConnectionFactory | None = None,
|
|
*,
|
|
connection: psycopg.Connection | None = None,
|
|
):
|
|
self._connection_factory = connection_factory or default_connection_factory
|
|
self._connection = connection
|
|
|
|
@contextmanager
|
|
def transaction(self) -> Iterator["InventoryRepository"]:
|
|
if self._connection is not None:
|
|
yield self
|
|
return
|
|
|
|
connection = self._connection_factory()
|
|
transaction_repository = InventoryRepository(
|
|
self._connection_factory,
|
|
connection=connection,
|
|
)
|
|
try:
|
|
yield transaction_repository
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
@contextmanager
|
|
def _connection_scope(
|
|
self,
|
|
*,
|
|
write: bool = False,
|
|
) -> Iterator[psycopg.Connection]:
|
|
if self._connection is not None:
|
|
yield self._connection
|
|
return
|
|
|
|
connection = self._connection_factory()
|
|
try:
|
|
yield connection
|
|
if write:
|
|
connection.commit()
|
|
except Exception:
|
|
if write:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def _fetch_miner(
|
|
cur,
|
|
where_sql: str,
|
|
params: Sequence[Any],
|
|
*,
|
|
for_update: bool = False,
|
|
) -> dict | None:
|
|
lock_sql = " FOR UPDATE OF miners" if for_update else ""
|
|
cur.execute(
|
|
f"{MINER_SELECT} WHERE {where_sql}{lock_sql}",
|
|
params,
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_miner_by_id(self, miner_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
return self._fetch_miner(cur, "miners.id = %s", (miner_id,))
|
|
|
|
def get_current_miner_by_id(self, miner_id: UUID) -> dict | None:
|
|
"""Return Inventory identity with its latest factual telemetry."""
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"{MINER_CURRENT_SELECT} WHERE miners.id = %s",
|
|
(miner_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_miner_by_id_for_update(self, miner_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
return self._fetch_miner(
|
|
cur,
|
|
"miners.id = %s",
|
|
(miner_id,),
|
|
for_update=True,
|
|
)
|
|
|
|
def get_miner_by_asset_id(self, asset_id: str) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
return self._fetch_miner(
|
|
cur,
|
|
"lower(trim(miners.asset_tag)) = lower(trim(%s))",
|
|
(asset_id,),
|
|
)
|
|
|
|
def list_miners(
|
|
self,
|
|
*,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
include_location_details: bool = False,
|
|
) -> list[dict]:
|
|
return self.search_miners(
|
|
limit=limit,
|
|
offset=offset,
|
|
include_location_details=include_location_details,
|
|
)
|
|
|
|
def summarize_fleet(self) -> dict[str, Any]:
|
|
"""Return exact fleet-wide counts using Inventory state semantics."""
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
WITH fleet_state AS (
|
|
SELECT miners.status, miners.lifecycle_status,
|
|
({OPERATIONAL_STATE_SQL}) AS operational_state
|
|
FROM miners
|
|
LEFT JOIN LATERAL (
|
|
SELECT snapshots.hash_stable
|
|
FROM miner_status_snapshots AS snapshots
|
|
WHERE snapshots.miner_id = miners.id
|
|
ORDER BY snapshots.observed_at DESC, snapshots.id DESC
|
|
LIMIT 1
|
|
) AS latest ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT repairs.id
|
|
FROM repairs
|
|
WHERE repairs.miner_id = miners.id
|
|
AND repairs.status = ANY(%s)
|
|
ORDER BY repairs.opened_at, repairs.id
|
|
LIMIT 1
|
|
) AS active_repair ON true
|
|
)
|
|
SELECT count(*)::integer AS total,
|
|
count(*) FILTER (
|
|
WHERE lower(trim(status)) = 'online'
|
|
)::integer AS runtime_online,
|
|
count(*) FILTER (
|
|
WHERE lower(trim(status)) = 'offline'
|
|
)::integer AS runtime_offline,
|
|
count(*) FILTER (
|
|
WHERE lower(trim(status)) NOT IN ('online', 'offline')
|
|
OR status IS NULL
|
|
)::integer AS runtime_unknown,
|
|
count(*) FILTER (
|
|
WHERE lifecycle_status = 'in_repair'
|
|
)::integer AS lifecycle_in_repair,
|
|
count(*) FILTER (
|
|
WHERE lifecycle_status = 'spare'
|
|
)::integer AS lifecycle_spare,
|
|
count(*) FILTER (
|
|
WHERE lifecycle_status = 'retired'
|
|
)::integer AS lifecycle_retired,
|
|
count(*) FILTER (
|
|
WHERE operational_state = 'offline'
|
|
)::integer AS attention_offline,
|
|
count(*) FILTER (
|
|
WHERE operational_state = 'unknown'
|
|
)::integer AS attention_unknown,
|
|
count(*) FILTER (
|
|
WHERE operational_state = 'unstable'
|
|
)::integer AS attention_unstable,
|
|
count(*) FILTER (
|
|
WHERE operational_state = 'repair'
|
|
)::integer AS attention_repair
|
|
FROM fleet_state
|
|
""",
|
|
(list(ACTIVE_REPAIR_STATUSES),),
|
|
)
|
|
return dict(cur.fetchone())
|
|
|
|
def search_miners(
|
|
self,
|
|
*,
|
|
query: str | None = None,
|
|
asset_id: str | None = None,
|
|
lifecycle_status: str | None = None,
|
|
model: str | None = None,
|
|
runtime_status: str | None = None,
|
|
repair: str | None = None,
|
|
operational_state: str | None = None,
|
|
problems: bool = False,
|
|
site_id: UUID | None = None,
|
|
room_id: UUID | None = None,
|
|
tag_names: Sequence[str] = (),
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
include_location_details: bool = False,
|
|
) -> list[dict]:
|
|
conditions: list[str] = []
|
|
params: list[Any] = [list(ACTIVE_REPAIR_STATUSES)]
|
|
|
|
if query:
|
|
pattern = f"%{query}%"
|
|
conditions.append(
|
|
"""
|
|
(
|
|
miners.asset_tag ILIKE %s
|
|
OR miners.current_mac ILIKE %s
|
|
OR host(miners.current_ip) ILIKE %s
|
|
OR miners.model ILIKE %s
|
|
OR miners.miner_sn ILIKE %s
|
|
OR miners.control_board_sn ILIKE %s
|
|
OR miners.power_sn ILIKE %s
|
|
OR miners.notes ILIKE %s
|
|
)
|
|
"""
|
|
)
|
|
params.extend([pattern] * 8)
|
|
if asset_id:
|
|
conditions.append("lower(trim(miners.asset_tag)) = lower(trim(%s))")
|
|
params.append(asset_id)
|
|
if lifecycle_status:
|
|
conditions.append("miners.lifecycle_status = %s")
|
|
params.append(lifecycle_status)
|
|
if model:
|
|
conditions.append("lower(trim(miners.model)) = lower(trim(%s))")
|
|
params.append(model)
|
|
if runtime_status:
|
|
conditions.append("lower(trim(miners.status)) = lower(trim(%s))")
|
|
params.append(runtime_status)
|
|
if repair == "active":
|
|
conditions.append("active_repair.id IS NOT NULL")
|
|
elif repair == "none":
|
|
conditions.append("active_repair.id IS NULL")
|
|
if operational_state:
|
|
conditions.append(f"({OPERATIONAL_STATE_SQL}) = %s")
|
|
params.append(operational_state)
|
|
if problems:
|
|
conditions.append(
|
|
f"({OPERATIONAL_STATE_SQL}) = ANY(%s)"
|
|
)
|
|
params.append(["offline", "unknown", "unstable", "repair"])
|
|
if site_id:
|
|
conditions.append("mapped.site_id = %s")
|
|
params.append(site_id)
|
|
if room_id:
|
|
conditions.append("mapped.room_id = %s")
|
|
params.append(room_id)
|
|
for tag_name in tag_names:
|
|
conditions.append(
|
|
f"""
|
|
EXISTS (
|
|
SELECT 1
|
|
FROM miner_tags search_miner_tags
|
|
JOIN tags search_tags ON search_tags.id = search_miner_tags.tag_id
|
|
WHERE search_miner_tags.miner_id = miners.id
|
|
AND lower({TAG_TRIMMED_NAME_SQL.replace("name", "search_tags.name")})
|
|
= %s
|
|
)
|
|
"""
|
|
)
|
|
params.append(tag_name)
|
|
|
|
location_fields = (
|
|
LOCATION_FIELDS if include_location_details else REDACTED_LOCATION_FIELDS
|
|
)
|
|
location_join = LOCATION_JOIN if include_location_details else ""
|
|
select_sql = FLEET_SELECT.format(
|
|
location_fields=location_fields,
|
|
location_join=location_join,
|
|
)
|
|
where_sql = " AND ".join(conditions) if conditions else "true"
|
|
params.extend([limit, offset])
|
|
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
{select_sql}
|
|
WHERE {where_sql}
|
|
ORDER BY lower(trim(miners.asset_tag)) NULLS LAST,
|
|
miners.created_at, miners.id
|
|
LIMIT %s OFFSET %s
|
|
""",
|
|
params,
|
|
)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
def list_location_filter_options(self) -> dict[str, list[dict]]:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT sites.id, sites.name
|
|
FROM location_positions lp
|
|
JOIN miners ON miners.id = lp.miner_id
|
|
JOIN rooms ON rooms.id = lp.room_id
|
|
JOIN sites ON sites.id = rooms.site_id
|
|
GROUP BY sites.id, sites.name
|
|
ORDER BY lower(sites.name), sites.id
|
|
"""
|
|
)
|
|
sites = [dict(row) for row in cur.fetchall()]
|
|
cur.execute(
|
|
"""
|
|
SELECT rooms.id, rooms.site_id, rooms.name
|
|
FROM location_positions lp
|
|
JOIN miners ON miners.id = lp.miner_id
|
|
JOIN rooms ON rooms.id = lp.room_id
|
|
GROUP BY rooms.id, rooms.site_id, rooms.name
|
|
ORDER BY lower(rooms.name), rooms.id
|
|
"""
|
|
)
|
|
rooms = [dict(row) for row in cur.fetchall()]
|
|
return {"sites": sites, "rooms": rooms}
|
|
|
|
def create_miner(self, values: Mapping[str, Any]) -> dict:
|
|
clean_values = {
|
|
key: value for key, value in values.items() if key in MINER_WRITE_COLUMNS
|
|
}
|
|
columns = list(clean_values)
|
|
params = [clean_values[column] for column in columns]
|
|
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
if columns:
|
|
placeholders = ", ".join(["%s"] * len(columns))
|
|
cur.execute(
|
|
f"""
|
|
INSERT INTO miners ({", ".join(columns)})
|
|
VALUES ({placeholders})
|
|
RETURNING id
|
|
""",
|
|
params,
|
|
)
|
|
else:
|
|
cur.execute("INSERT INTO miners DEFAULT VALUES RETURNING id")
|
|
miner_id = cur.fetchone()["id"]
|
|
return self._fetch_miner(cur, "miners.id = %s", (miner_id,))
|
|
|
|
def update_miner(self, miner_id: UUID, values: Mapping[str, Any]) -> dict | None:
|
|
clean_values = {
|
|
key: value for key, value in values.items() if key in MINER_WRITE_COLUMNS
|
|
}
|
|
if not clean_values:
|
|
return self.get_miner_by_id(miner_id)
|
|
|
|
assignments = [f"{column} = %s" for column in clean_values]
|
|
params = list(clean_values.values())
|
|
params.append(miner_id)
|
|
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
UPDATE miners
|
|
SET {", ".join(assignments)}, updated_at = now()
|
|
WHERE id = %s
|
|
RETURNING id
|
|
""",
|
|
params,
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
return self._fetch_miner(cur, "miners.id = %s", (miner_id,))
|
|
|
|
def add_asset_id_history(
|
|
self,
|
|
miner_id: UUID,
|
|
old_asset_id: str | None,
|
|
new_asset_id: str,
|
|
changed_by: UUID,
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_asset_id_history (
|
|
miner_id,
|
|
old_asset_id,
|
|
new_asset_id,
|
|
changed_by
|
|
)
|
|
VALUES (%s, %s, %s, %s)
|
|
RETURNING
|
|
id,
|
|
miner_id,
|
|
old_asset_id,
|
|
new_asset_id,
|
|
changed_at,
|
|
changed_by
|
|
""",
|
|
(miner_id, old_asset_id, new_asset_id, changed_by),
|
|
)
|
|
return dict(cur.fetchone())
|
|
|
|
def list_asset_id_history(self, miner_id: UUID) -> list[dict]:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
history.id,
|
|
history.miner_id,
|
|
history.old_asset_id,
|
|
history.new_asset_id,
|
|
history.changed_at,
|
|
history.changed_by,
|
|
changed_by_user.display_name
|
|
AS changed_by_display_name
|
|
FROM miner_asset_id_history AS history
|
|
JOIN users AS changed_by_user
|
|
ON changed_by_user.id = history.changed_by
|
|
WHERE history.miner_id = %s
|
|
ORDER BY history.changed_at, history.id
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
def list_tags(self) -> list[dict]:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT id, name, created_at, updated_at
|
|
FROM tags
|
|
ORDER BY lower({TAG_TRIMMED_NAME_SQL}), id
|
|
"""
|
|
)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
def add_tag(self, miner_id: UUID, tag_name: str) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
INSERT INTO tags (name)
|
|
VALUES (%s)
|
|
ON CONFLICT (lower({TAG_TRIMMED_NAME_SQL}))
|
|
DO UPDATE SET updated_at = tags.updated_at
|
|
RETURNING id, name, created_at, updated_at
|
|
""",
|
|
(tag_name,),
|
|
)
|
|
tag = dict(cur.fetchone())
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_tags (miner_id, tag_id)
|
|
VALUES (%s, %s)
|
|
ON CONFLICT (miner_id, tag_id) DO NOTHING
|
|
""",
|
|
(miner_id, tag["id"]),
|
|
)
|
|
return tag
|
|
|
|
def remove_tag(self, miner_id: UUID, tag_id: UUID) -> bool:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"DELETE FROM miner_tags WHERE miner_id = %s AND tag_id = %s",
|
|
(miner_id, tag_id),
|
|
)
|
|
return cur.rowcount > 0
|