130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# Libervia plugin to handle forges.
|
|
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# 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 Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
from typing import Iterator
|
|
|
|
from pydantic import BaseModel, Field, RootModel
|
|
from twisted.internet import defer
|
|
|
|
from libervia.backend import G
|
|
from libervia.backend.core import exceptions
|
|
from libervia.backend.core.constants import Const as C
|
|
from libervia.backend.core.core_types import SatXMPPEntity
|
|
from libervia.backend.core.i18n import _
|
|
from libervia.backend.core.log import getLogger
|
|
from libervia.backend.models.types import JIDType
|
|
from libervia.backend.tools.config import ConfigOptions, DirectoryPath, PluginSettings
|
|
|
|
log = getLogger(__name__)
|
|
|
|
|
|
PLUGIN_INFO = {
|
|
C.PI_NAME: "Forge",
|
|
C.PI_IMPORT_NAME: "FORGE",
|
|
C.PI_TYPE: C.PLUG_TYPE_MISC,
|
|
C.PI_DEPENDENCIES: [],
|
|
C.PI_MAIN: "Forge",
|
|
C.PI_HANDLER: "no",
|
|
C.PI_DESCRIPTION: _("""Forge for developping projects."""),
|
|
}
|
|
|
|
|
|
class TicketsNode(ConfigOptions):
|
|
service: JIDType
|
|
node: str
|
|
|
|
|
|
class Repository(ConfigOptions):
|
|
name: str
|
|
path: DirectoryPath
|
|
tickets: TicketsNode | None = None
|
|
|
|
|
|
class Repositories(RootModel):
|
|
root: list[Repository]
|
|
|
|
def __iter__(self) -> Iterator[Repository]: # type: ignore
|
|
return iter(self.root)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.root)
|
|
|
|
|
|
class ProfileRepositories(RootModel):
|
|
root: dict[str, Repositories]
|
|
|
|
def __getitem__(self, item: str) -> Repositories:
|
|
return self.root[item]
|
|
|
|
|
|
class DefaultRepository(BaseModel):
|
|
profile: str
|
|
repository: str
|
|
|
|
|
|
class PluginForgeSettings(PluginSettings, plugin_info=PLUGIN_INFO):
|
|
repositories: ProfileRepositories = Field(
|
|
default_factory=lambda: ProfileRepositories({})
|
|
)
|
|
default: DefaultRepository | None = None
|
|
|
|
|
|
class Forge:
|
|
def __init__(self, host):
|
|
log.info(f"Plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization.")
|
|
self.settings = G.config.get_settings(PluginForgeSettings)
|
|
self.host = host
|
|
|
|
host.bridge.add_method(
|
|
"forge_repositories_get",
|
|
".plugin",
|
|
in_sign="sss",
|
|
out_sign="s",
|
|
method=self._repositories_get,
|
|
async_=True,
|
|
)
|
|
host.bridge.add_method(
|
|
"forge_default_repository_get",
|
|
".plugin",
|
|
in_sign="",
|
|
out_sign="s",
|
|
method=self._default_repository_get,
|
|
async_=True,
|
|
)
|
|
|
|
def _repositories_get(
|
|
self, target_profile: str, data_s: str, profile: str
|
|
) -> defer.Deferred[str]:
|
|
client = self.host.get_client(profile)
|
|
repositories = self.repositories_get(client, target_profile)
|
|
return defer.succeed(repositories.model_dump_json())
|
|
|
|
def repositories_get(
|
|
self, client: SatXMPPEntity, target_profile: str, data: dict | None = None
|
|
) -> Repositories:
|
|
try:
|
|
return self.settings.repositories[target_profile]
|
|
except KeyError:
|
|
raise exceptions.NotFound(f"No repository found for {target_profile!r}.")
|
|
|
|
def _default_repository_get(self) -> defer.Deferred[str]:
|
|
default = self.settings.default
|
|
if default is None:
|
|
raise exceptions.NotFound("No default repository set.")
|
|
else:
|
|
return defer.succeed(default.model_dump_json())
|