JiriUhlir
122 lines
2.3 KiB
Python
122 lines
2.3 KiB
Python
from app.db.database import get_connection
|
|
|
|
|
|
def get_apps():
|
|
con = get_connection()
|
|
|
|
rows = con.execute(
|
|
"""
|
|
SELECT
|
|
a.id,
|
|
a.name,
|
|
a.language,
|
|
a.version,
|
|
a.status,
|
|
COALESCE(a.memory, '') AS memory,
|
|
COALESCE(a.cpus, '') AS cpus,
|
|
a.updated_at,
|
|
(
|
|
SELECT d.status
|
|
FROM deployments d
|
|
WHERE d.app_id = a.id
|
|
ORDER BY d.id DESC
|
|
LIMIT 1
|
|
) AS last_deploy_status,
|
|
(
|
|
SELECT d.started_at
|
|
FROM deployments d
|
|
WHERE d.app_id = a.id
|
|
ORDER BY d.id DESC
|
|
LIMIT 1
|
|
) AS last_deploy_started_at
|
|
FROM apps a
|
|
ORDER BY a.id
|
|
"""
|
|
).fetchall()
|
|
|
|
con.close()
|
|
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def get_app(app_id: str):
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
name,
|
|
language,
|
|
version,
|
|
status,
|
|
COALESCE(memory, '') AS memory,
|
|
COALESCE(cpus, '') AS cpus,
|
|
updated_at
|
|
FROM apps
|
|
WHERE id = ?
|
|
""",
|
|
(app_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return dict(row)
|
|
|
|
|
|
def get_deployments(limit: int = 100):
|
|
con = get_connection()
|
|
|
|
rows = con.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
app_id,
|
|
kind,
|
|
status,
|
|
started_at,
|
|
finished_at,
|
|
returncode
|
|
FROM deployments
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
|
|
con.close()
|
|
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def get_deployment(deployment_id: int):
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
app_id,
|
|
kind,
|
|
status,
|
|
started_at,
|
|
finished_at,
|
|
returncode,
|
|
stdout,
|
|
stderr
|
|
FROM deployments
|
|
WHERE id = ?
|
|
""",
|
|
(deployment_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return dict(row)
|