ryan
68 lines
1.7 KiB
Python
68 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
|
|
|
|
def sql_literal(value: object) -> str:
|
|
if value is None:
|
|
return "NULL"
|
|
if isinstance(value, bool):
|
|
return "TRUE" if value else "FALSE"
|
|
text = str(value)
|
|
return "'" + text.replace("'", "''") + "'"
|
|
|
|
|
|
@dataclass
|
|
class Database:
|
|
url: str
|
|
|
|
def execute(self, sql: str) -> None:
|
|
subprocess.run(
|
|
["psql", self.url, "-X", "-v", "ON_ERROR_STOP=1", "-q"],
|
|
input=sql,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
def rows(self, sql: str) -> list[dict[str, str | None]]:
|
|
result = subprocess.run(
|
|
[
|
|
"psql",
|
|
self.url,
|
|
"-X",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
"-q",
|
|
"-P",
|
|
"footer=off",
|
|
"-A",
|
|
"-F",
|
|
"\t",
|
|
"-c",
|
|
sql,
|
|
],
|
|
text=True,
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
lines = [line for line in result.stdout.splitlines() if line.strip()]
|
|
if not lines:
|
|
return []
|
|
headers = lines[0].split("\t")
|
|
out: list[dict[str, str | None]] = []
|
|
for line in lines[1:]:
|
|
values = line.split("\t")
|
|
out.append(
|
|
{
|
|
key: None if idx >= len(values) or values[idx] == "" else values[idx]
|
|
for idx, key in enumerate(headers)
|
|
}
|
|
)
|
|
return out
|
|
|
|
def one(self, sql: str) -> dict[str, str | None] | None:
|
|
rows = self.rows(sql)
|
|
return rows[0] if rows else None
|
|
|