Source code for ferrosoft.ops.models

import datetime

from attrs import define, field, Factory

from ferrosoft.ops import sql




@define
class Tenant:
    id: str
    name: str
    db_name: str
    db_hostname: str
    db_port: str
    db_username: str
    db_password: str





@define
class JournalEntry:
    filename: str
    pg_dump_args: str
    db_name: str | None
    tenant: Tenant | None
    filename_decompressed: str | None = None





@define
class Journal:
    backup_date: datetime.datetime
    entries: list[JournalEntry] = field(default=Factory(list))



    def master_entry(self) -> JournalEntry | None:
        for entry in self.entries:
            if entry.tenant is None:
                return entry
        return None




    def tenant_entry(self, tenant_id: str) -> JournalEntry | None:
        for entry in self.entries:
            if entry.tenant is not None and entry.tenant.id == tenant_id:
                return entry
        return None




# noinspection SqlResolve


def get_tenants(psql_executable: str):
    for row in sql.sql_select(
        "SELECT id, name, database_name, database_hostname, database_port, "
        "database_username,database_password FROM ferrobase_tenant",
        psql_executable,
    ):
        yield Tenant(
            id=row[0],
            name=row[1],
            db_name=row[2],
            db_hostname=row[3],
            db_port=row[4],
            db_username=row[5],
            db_password=row[6],
        )





def get_roles(psql_executable: str) -> list[str]:
    return [row[0] for row in sql.sql_select("\\dg", psql_executable)]





def get_session_user(psql_executable: str) -> str:
    return list(sql.sql_select("SELECT session_user", psql_executable))[0][0]



# noinspection SqlResolve


def get_databases(psql_executable) -> list[str]:
    return [
        row[0]
        for row in sql.sql_select("SELECT datname FROM pg_database", psql_executable)
    ]