2023-10-24 20:44:48 +00:00
|
|
|
# Copyright © 2022, 2023 Vulcanize
|
|
|
|
|
|
|
|
# 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 pathlib import Path
|
|
|
|
import typing
|
2023-11-07 07:06:55 +00:00
|
|
|
from stack_orchestrator.util import get_yaml
|
2023-11-21 23:04:36 +00:00
|
|
|
from stack_orchestrator import constants
|
2023-10-24 20:44:48 +00:00
|
|
|
|
|
|
|
|
|
|
|
class Spec:
|
|
|
|
|
|
|
|
obj: typing.Any
|
2024-01-31 05:09:48 +00:00
|
|
|
file_path: Path
|
2023-10-24 20:44:48 +00:00
|
|
|
|
|
|
|
def __init__(self) -> None:
|
|
|
|
pass
|
|
|
|
|
|
|
|
def init_from_file(self, file_path: Path):
|
|
|
|
with file_path:
|
|
|
|
self.obj = get_yaml().load(open(file_path, "r"))
|
2024-01-31 05:09:48 +00:00
|
|
|
self.file_path = file_path
|
2023-11-21 23:04:36 +00:00
|
|
|
|
|
|
|
def get_image_registry(self):
|
|
|
|
return (self.obj[constants.image_resigtry_key]
|
|
|
|
if self.obj and constants.image_resigtry_key in self.obj
|
|
|
|
else None)
|
|
|
|
|
2024-01-31 05:09:48 +00:00
|
|
|
def get_volumes(self):
|
|
|
|
return (self.obj["volumes"]
|
|
|
|
if self.obj and "volumes" in self.obj
|
|
|
|
else {})
|
|
|
|
|
|
|
|
def get_configmaps(self):
|
|
|
|
return (self.obj["configmaps"]
|
|
|
|
if self.obj and "configmaps" in self.obj
|
|
|
|
else {})
|
|
|
|
|
2024-02-07 22:48:02 +00:00
|
|
|
def get_container_resources(self):
|
|
|
|
return self.obj.get("resources", {}).get("containers")
|
|
|
|
|
|
|
|
def get_volume_resources(self):
|
|
|
|
return self.obj.get("resources", {}).get("volumes")
|
|
|
|
|
2023-11-21 23:04:36 +00:00
|
|
|
def get_http_proxy(self):
|
|
|
|
return (self.obj[constants.network_key][constants.http_proxy_key]
|
|
|
|
if self.obj and constants.network_key in self.obj
|
|
|
|
and constants.http_proxy_key in self.obj[constants.network_key]
|
|
|
|
else None)
|