Skip to content

Commit

Permalink
Changes to support Pydantic v2
Browse files Browse the repository at this point in the history
  • Loading branch information
mkjpryor committed Nov 8, 2023
1 parent 1a46517 commit bf0ed97
Show file tree
Hide file tree
Showing 7 changed files with 156 additions and 162 deletions.
95 changes: 52 additions & 43 deletions azimuth_capi/config.py
Original file line number Diff line number Diff line change
@@ -1,27 +1,32 @@
import typing as t

from pydantic import Field, AnyHttpUrl, FilePath, conint, constr, root_validator, validator
from pydantic import (
TypeAdapter,
Field,
AfterValidator,
StringConstraints,
AnyHttpUrl as PyAnyHttpUrl,
FilePath,
conint,
constr,
model_validator,
field_validator,
ValidationInfo
)

from configomatic import Configuration as BaseConfiguration, Section, LoggingConfiguration

from easysemver import Version
from easysemver import SEMVER_VERSION_REGEX


class SemVerVersion(str):
"""
Type for a string that is a valid SemVer version.
"""
@classmethod
def __get_validators__(cls):
yield cls.validate

@classmethod
def validate(cls, v):
# Just use the validation from easysemver, then convert to a string
return cls(Version(v))
#: Type for a string that validates as a SemVer version
SemVerVersion = t.Annotated[str, StringConstraints(pattern = SEMVER_VERSION_REGEX)]

def __repr__(self):
return f'{self.__class__.__name__}({super().__repr__()})'
#: Type for a string that validates as a URL
AnyHttpUrl = t.Annotated[
str,
AfterValidator(lambda v: TypeAdapter(PyAnyHttpUrl).validate_python(v))
]


class HelmClientConfiguration(Section):
Expand Down Expand Up @@ -67,7 +72,7 @@ class ZenithConfig(Section):
registrar_admin_url: t.Optional[AnyHttpUrl] = None
#: The internal admin URL of the Zenith registrar
#: By default, this is the same as the registrar_admin_url
registrar_admin_url_internal: t.Optional[AnyHttpUrl] = None
registrar_admin_url_internal: t.Optional[AnyHttpUrl] = Field(None, validate_default = True)
#: The host for the Zenith SSHD service
sshd_host: t.Optional[constr(min_length = 1)] = None
#: The port for the Zenith SSHD service
Expand All @@ -90,26 +95,27 @@ class ZenithConfig(Section):
monitoring_icon_url: AnyHttpUrl = "https://raw.githubusercontent.com/cncf/artwork/master/projects/prometheus/icon/color/prometheus-icon-color.png"

#: The API version to use when watching Zenith resources on target clusters
api_version: constr(regex = r"^[a-z0-9.-]+/[a-z0-9]+$") = "zenith.stackhpc.com/v1alpha1"
api_version: constr(pattern =r"^[a-z0-9.-]+/[a-z0-9]+$") = "zenith.stackhpc.com/v1alpha1"

@root_validator
def validate_zenith_enabled(cls, values):
@model_validator(mode = "after")
def validate_zenith_enabled(self):
"""
Ensures that the SSHD host is set when the registrar URL is given.
"""
if bool(values.get("registrar_admin_url")) != bool(values.get("sshd_host")):
if bool(self.registrar_admin_url) != bool(self.sshd_host):
raise ValueError(
"registrar_admin_url and sshd_host are both required to "
"enable Zenith support"
)
return values
return self

@validator("registrar_admin_url_internal", always = True)
def default_registrar_admin_url_internal(cls, v, values):
@field_validator("registrar_admin_url_internal")
@classmethod
def default_registrar_admin_url_internal(cls, v, info: ValidationInfo):
"""
Sets the default internal registrar admin URL.
"""
return v or values.get("registrar_admin_url")
return v or info.data.get("registrar_admin_url")

@property
def enabled(self):
Expand Down Expand Up @@ -140,49 +146,52 @@ class WebhookConfiguration(Section):
#: Indicates whether kopf should manage the webhook configurations
managed: bool = False
#: The path to the TLS certificate to use
certfile: t.Optional[FilePath] = None
certfile: t.Optional[FilePath] = Field(None, validate_default = True)
#: The path to the key for the TLS certificate
keyfile: t.Optional[FilePath] = None
keyfile: t.Optional[FilePath] = Field(None, validate_default = True)
#: The host for the webhook server (required for self-signed certificate generation)
host: t.Optional[constr(min_length = 1)] = None
host: t.Optional[constr(min_length = 1)] = Field(None, validate_default = True)

@validator("certfile", always = True)
def validate_certfile(cls, v, values, **kwargs):
@field_validator("certfile")
@classmethod
def validate_certfile(cls, v, info: ValidationInfo):
"""
Validate that certfile is specified when configs are not managed.
"""
if "managed" in values and not values["managed"] and v is None:
if not info.data.get("managed") and v is None:
raise ValueError("required when webhook configurations are not managed")
return v

@validator("keyfile", always = True)
def validate_keyfile(cls, v, values, **kwargs):
@field_validator("keyfile")
@classmethod
def validate_keyfile(cls, v, info: ValidationInfo):
"""
Validate that keyfile is specified when certfile is present.
"""
if "certfile" in values and values["certfile"] is not None and v is None:
if info.data.get("certfile") is not None and v is None:
raise ValueError("required when certfile is given")
return v

@validator("host", always = True)
def validate_host(cls, v, values, **kwargs):
@field_validator("host")
@classmethod
def validate_host(cls, v, info: ValidationInfo):
"""
Validate that host is specified when there is no certificate specified.
"""
if values.get("certfile") is None and v is None:
if info.data.get("certfile") is None and v is None:
raise ValueError("required when certfile is not given")
return v


class Configuration(BaseConfiguration):
class Configuration(
BaseConfiguration,
default_path = "/etc/azimuth/capi-operator.yaml",
path_env_var = "AZIMUTH_CAPI_CONFIG",
env_prefix = "AZIMUTH_CAPI"
):
"""
Top-level configuration model.
"""
class Config:
default_path = "/etc/azimuth/capi-operator.yaml"
path_env_var = "AZIMUTH_CAPI_CONFIG"
env_prefix = "AZIMUTH_CAPI"

#: The logging configuration
logging: LoggingConfiguration = Field(default_factory = LoggingConfiguration)

Expand Down
29 changes: 10 additions & 19 deletions azimuth_capi/models/v1alpha1/app_template.py
Original file line number Diff line number Diff line change
@@ -1,28 +1,22 @@
import datetime as dt
import typing as t

from pydantic import Extra, Field, AnyHttpUrl, constr
from pydantic import Field

from kube_custom_resource import CustomResource, Scope, schema

from easysemver import SEMVER_VERSION_REGEX, SEMVER_RANGE_REGEX


#: Type for a SemVer version
SemVerVersion = constr(regex = SEMVER_VERSION_REGEX)
#: Type for a SemVer range
SemVerRange = constr(regex = SEMVER_RANGE_REGEX)


class AppTemplateChartSpec(schema.BaseModel):
"""
The spec for a chart reference for a Kubernetes app template.
"""
repo: AnyHttpUrl = Field(
repo: schema.AnyHttpUrl = Field(
...,
description = "The Helm repository that the chart is in."
)
name: constr(regex = r"^[a-z0-9-]+$") = Field(
name: schema.constr(pattern =r"^[a-z0-9-]+$") = Field(
...,
description = "The name of the chart."
)
Expand Down Expand Up @@ -57,7 +51,7 @@ class AppTemplateSpec(schema.BaseModel):
"If not given, the description from the Chart.yaml of the chart will be used."
)
)
version_range: SemVerRange = Field(
version_range: schema.constr(pattern = SEMVER_RANGE_REGEX) = Field(
">=0.0.0",
description = (
"The range of versions to make available. "
Expand Down Expand Up @@ -90,7 +84,7 @@ class AppTemplateVersion(schema.BaseModel):
"""
The available versions for the app template.
"""
name: SemVerVersion = Field(
name: schema.constr(pattern = SEMVER_VERSION_REGEX) = Field(
...,
description = "The name of the version."
)
Expand All @@ -104,30 +98,27 @@ class AppTemplateVersion(schema.BaseModel):
)


class AppTemplateStatus(schema.BaseModel):
class AppTemplateStatus(schema.BaseModel, extra = "allow"):
"""
The status of the app template.
"""
class Config:
extra = Extra.allow

label: t.Optional[constr(min_length = 1)] = Field(
label: schema.Optional[schema.constr(min_length = 1)] = Field(
None,
description = "The human-readable label for the app template."
)
logo: t.Optional[constr(min_length = 1)] = Field(
logo: schema.Optional[schema.constr(min_length = 1)] = Field(
None,
description = "The URL of the logo for the app template."
)
description: t.Optional[constr(min_length = 1)] = Field(
description: schema.Optional[schema.constr(min_length = 1)] = Field(
None,
description = "A short description of the app template."
)
versions: t.List[AppTemplateVersion] = Field(
default_factory = list,
description = "The available versions for the app template."
)
last_sync: t.Optional[dt.datetime] = Field(
last_sync: schema.Optional[dt.datetime] = Field(
None,
description = "The time that the last successful sync of versions took place."
)
Expand Down
Loading

0 comments on commit bf0ed97

Please sign in to comment.