path: root/buildbot_gentoo_ci/db/versions.py

# This file has parts from Buildbot and is modifyed by Gentoo Authors. 
# Buildbot is free software: you can redistribute it and/or modify it 
# under the terms of the GNU General Public License as published by the 
# Free Software Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members
# Origins: buildbot.db.*
# Modifyed by Gentoo Authors.
# Copyright 2024 Gentoo Authors

from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING

import uuid
import sqlalchemy as sa

from twisted.internet import defer

from buildbot.db import base
from buildbot.warnings import warn_deprecated

if TYPE_CHECKING:
    import datetime

@dataclass
class VersionModel:
    uuid: str
    name: str
    package_uuid: str
    file_hash: str
    commit_id: str
    change_id: int
    created_at: datetime.datetime | None
    deleted: bool
    deleted_at: datetime.datetime | None

    # For backward compatibility
    def __getitem__(self, key: str):
        if hasattr(self, key):
            return getattr(self, key)
        raise KeyError(key)

def _db2data_Version(model: VersionModel):
    if model is None:
        return None
    return {
            'uuid' : model.uuid,
            'name' : model.name,
            'package_uuid' : model.package_uuid,
            'file_hash' : model.file_hash,
            'commit_id' : model.commit_id,
            'change_id' : model.change_id,
            'deleted' : model.deleted,
            'deleted_at' : model.deleted_at,
            'created_at' : model.created_at,
    }

class VersionsConnectorComponent(base.DBConnectorComponent):

    def getVersionByName(self, name: str, p_uuid: str, deleted: bool | None = False) -> defer.Deferred[VersionModel | None]:
        def thd(conn) -> VersionModel | None:
            tbl = self.db.model.versions
            q = tbl.select()
            q = q.where(tbl.c.name == name)
            q = q.where(tbl.c.package_uuid == p_uuid)
            q = q.where(tbl.c.deleted == deleted)
            res = conn.execute(q)
            row = res.fetchone()
            rv = None
            if row:
                rv = self._model_from_row_VersionModel(row)
            res.close()
            return rv
        return self.db.pool.do(thd)

    def getVersionByUuid(self, uuid: str) -> defer.Deferred[VersionModel | None]:
        def thd(conn) -> VersionModel | None:
            tbl = self.db.model.versions
            q = tbl.select()
            q = q.where(tbl.c.uuid == uuid)
            res = conn.execute(q)
            row = res.fetchone()
            rv = None
            if row:
                rv = self._model_from_row_VersionModel(row)
            res.close()
            return rv
        return self.db.pool.do(thd)

    def addVersion(self, name: str, package_uuid: str, file_hash: str, commit_id: int, change_id: int) -> defer.Deferred[str]:
        def thd(conn) -> str:
            insert_row = {
                'name': name,
                'package_uuid' : package_uuid,
                'file_hash' : file_hash,
                'commit_id' : commit_id,
                'change_id' : change_id,
                'created_at' : int(self.master.reactor.seconds()),
                'deleted_at' : 0,
                'deleted' : False,
            }
            try:
                r = conn.execute(self.db.model.versions.insert(), insert_row)
                conn.commit()
            except (sa.exc.IntegrityError, sa.exc.ProgrammingError):
                conn.rollback()
                return None
            else:
                return r.inserted_primary_key[0]
        return self.db.pool.do(thd)

    def delVersion(self, uuid):
        def thd(conn):
            tbl = self.db.model.versions
            q = tbl.update().where(tbl.c.uuid == uuid)
            res = conn.execute(q.values(deleted=True, deleted_at=int(self.master.reactor.seconds())))
            #conn.commit()
            res.close()
        return self.db.pool.do(thd)

    @defer.inlineCallbacks
    def addKeyword(self, version_uuid, keyword_id, status):
        def thd(conn, no_recurse=False):
            try:
                tbl = self.db.model.versions_keywords
                q = tbl.insert()
                r = conn.execute(q, dict(version_uuid=version_uuid,
                                         keyword_id=keyword_id,
                                         status=status))
                conn.commit()
            except (sa.exc.IntegrityError, sa.exc.ProgrammingError):
                uuid = None
            else:
                uuid = r.inserted_primary_key[0]
            return uuid
        res = yield self.db.pool.do(thd)
        return res

    @defer.inlineCallbacks
    def addMetadata(self, version_uuid, metadata, value):
        def thd(conn, no_recurse=False):
            try:
                tbl = self.db.model.versions_metadata
                q = tbl.insert()
                r = conn.execute(q, dict(version_uuid=version_uuid,
                                         metadata=metadata,
                                         value=value))
                #conn.commit()
            except (sa.exc.IntegrityError, sa.exc.ProgrammingError):
                id = None
            else:
                id = r.inserted_primary_key[0]
            return uuid
        res = yield self.db.pool.do(thd)
        return res

    #FIXME: return sorted by id
    @defer.inlineCallbacks
    def getMetadataByUuidAndMatadata(self, uuid, metadata):
        def thd(conn):
            tbl = self.db.model.versions_metadata
            q = tbl.select()
            q = q.where(tbl.c.version_uuid == uuid)
            q = q.where(tbl.c.metadata == metadata)
            return [self._row2dict_version_metadata(conn, row)
                for row in conn.execute(q).fetchall()]
        res = yield self.db.pool.do(thd)
        return res

    @defer.inlineCallbacks
    def getEbuildsByPackage(self, p_uuid, deleted=False):
        def thd(conn):
            tbl = self.db.model.versions
            q = tbl.select()
            q = q.where(tbl.c.deleted == deleted)
            q = q.where(tbl.c.package_uuid == p_uuid)
            return [self._model_from_row_VersionModel(row)
                for row in conn.execute(q).fetchall()]
        res = yield self.db.pool.do(thd)
        return res

    def removeVersion(self, uuid: str) -> defer.Deferred[None]:
        def thd(conn) -> None:
                tbl = self.db.model.versions
                q = tbl.delete()
                q = q.where(tbl.c.uuid == uuid)
                res = conn.execute(q)
                conn.commit()
                res.close()
        return self.db.pool.do(thd)

    def removeVersionMetadata(self, version_uuid: str) -> defer.Deferred[None]:
        def thd(conn) -> None:
                tbl = self.db.model.versions_metadata
                q = tbl.delete()
                q = q.where(tbl.c.version_uuid == version_uuid)
                res = conn.execute(q)
                conn.commit()
                res.close()
        return self.db.pool.do(thd)

    def removeVersionKeyword(self, version_uuid: str) -> defer.Deferred[None]:
        def thd(conn) -> None:
                tbl = self.db.model.versions_keywords
                q = tbl.delete()
                q = q.where(tbl.c.version_uuid == version_uuid)
                res = conn.execute(q)
                conn.commit()
                res.close()
        return self.db.pool.do(thd)


    def _model_from_row_VersionModel(self, row):
        return VersionModel(
            uuid=row.uuid,
            name=row.name,
            package_uuid=row.package_uuid,
            file_hash=row.file_hash,
            commit_id=row.commit_id,
            change_id = row.change_id,
            deleted=row.deleted,
            deleted_at=row.deleted_at,
            created_at=row.created_at,
            )

    def _row2dict_version_metadata(self, conn, row):
        return dict(
            id=row.id,
            version_uuid=row.version_uuid,
            metadata=row.metadata,
            value=row.value
            )