"""Path type definitions for SDP config entity models."""
from pathlib import PurePath
from typing import Annotated, Sequence
from pydantic import Field, PlainSerializer, PlainValidator, WrapValidator
from ska_sdp_config.entity.base import EntityBaseModel
STR_SERIALIZER = PlainSerializer(str, return_type=str)
PATH_JSON_SCHEMA = Field(
json_schema_extra={
"format": "path",
"type": "string",
}
)
def _validate_pure_path(value):
if not isinstance(value, PurePath):
value = PurePath(value)
return value
def _validate_is_absolute(value: PurePath, handler):
value = handler(value)
assert value.is_absolute(), f"'{value}' is not an absolute path"
return value
def _validate_is_relative(value: PurePath, handler):
value = handler(value)
assert not value.is_absolute(), f"'{value}' is an absolute path"
return value
AnyPurePath = Annotated[
PurePath,
PlainValidator(_validate_pure_path, str),
STR_SERIALIZER,
PATH_JSON_SCHEMA,
]
"""
A generic filesystem path. No filesystem checks are performed during
validation.
"""
AbsolutePurePath = Annotated[
AnyPurePath,
WrapValidator(_validate_is_absolute, str),
STR_SERIALIZER,
PATH_JSON_SCHEMA,
]
"""An absolute filesystem path. No filesystem checks are performed
during validation."""
RelativePurePath = Annotated[
AnyPurePath,
WrapValidator(_validate_is_relative, str),
STR_SERIALIZER,
PATH_JSON_SCHEMA,
]
"""A relative filesystem path. No filesystem checks are performed
during validation."""
[docs]
class PVCPath(EntityBaseModel):
"""Mounted path within a Kubernetes Persistent Volume Claim."""
k8s_namespaces: str | Sequence[str]
"""Kuberentes pvc namespaces."""
k8s_pvc_name: str
"""Kubernetes persistent volume claim name."""
pvc_mount_path: AbsolutePurePath
"""Mount path of the persistent volume claim."""
pvc_subpath: RelativePurePath
"""Subpath within the persistent volume claim."""
@property
def path(self) -> AbsolutePurePath:
"""The container mounted absolute path to subpath."""
return self.pvc_mount_path / self.pvc_subpath
def __str__(self) -> str:
return str(self.path)
def __truediv__(self, other):
return self.path / other