92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
import db
|
|
|
|
|
|
def get_version_with_files(version_id: int):
|
|
conn = db.get_conn()
|
|
version = conn.execute("SELECT app_id,channel,version,create_time FROM versions WHERE id=?", (version_id,)).fetchone()
|
|
files = conn.execute("SELECT path,sha256,size FROM version_files WHERE version_id=? ORDER BY id", (version_id,)).fetchall()
|
|
conn.close()
|
|
return version, files
|
|
|
|
|
|
def list_versions(app_id: str) -> list[dict]:
|
|
conn = db.get_conn()
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, version, channel, latest, client_protocol, create_time
|
|
FROM versions WHERE app_id=? ORDER BY create_time DESC
|
|
""",
|
|
(app_id,),
|
|
).fetchall()
|
|
conn.close()
|
|
return [
|
|
{
|
|
"id": row["id"],
|
|
"version": row["version"],
|
|
"channel": row["channel"],
|
|
"latest": bool(row["latest"]),
|
|
"client_protocol": int(row["client_protocol"] or 1),
|
|
"create_time": row["create_time"],
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def set_version_protocol(version_id: int, client_protocol: int) -> int:
|
|
conn = db.get_conn()
|
|
cur = conn.execute("UPDATE versions SET client_protocol=? WHERE id=?", (client_protocol, version_id))
|
|
conn.commit()
|
|
updated = cur.rowcount
|
|
conn.close()
|
|
return updated
|
|
|
|
|
|
def set_latest_version(version_id: int) -> str:
|
|
conn = db.get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
version_info = cur.execute("SELECT app_id, channel FROM versions WHERE id=?", (version_id,)).fetchone()
|
|
if not version_info:
|
|
return "not_found"
|
|
app_id, channel = version_info["app_id"], version_info["channel"]
|
|
cur.execute("UPDATE versions SET latest=0 WHERE app_id=? AND channel=?", (app_id, channel))
|
|
cur.execute("UPDATE versions SET latest=1 WHERE id=?", (version_id,))
|
|
conn.commit()
|
|
return "ok"
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_version_for_delete(version_id: int):
|
|
conn = db.get_conn()
|
|
row = conn.execute("SELECT app_id, channel, version, latest FROM versions WHERE id=?", (version_id,)).fetchone()
|
|
conn.close()
|
|
return row
|
|
|
|
|
|
def delete_version_and_promote(version_id: int, app_id: str, channel: str, was_latest: bool) -> str | None:
|
|
conn = db.get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM version_files WHERE version_id=?", (version_id,))
|
|
cur.execute("DELETE FROM versions WHERE id=?", (version_id,))
|
|
promoted_version = None
|
|
if was_latest:
|
|
replacement = cur.execute(
|
|
"SELECT id, version FROM versions WHERE app_id=? AND channel=? ORDER BY create_time DESC, id DESC LIMIT 1",
|
|
(app_id, channel),
|
|
).fetchone()
|
|
if replacement:
|
|
cur.execute("UPDATE versions SET latest=1 WHERE id=?", (replacement["id"],))
|
|
promoted_version = replacement["version"]
|
|
conn.commit()
|
|
return promoted_version
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|