ipam.py

# mypy: ignore-errors
import ipaddress
from typing import Optional, Union

from pydantic import BaseSettings

from gso.services import _ipam
from gso.services._ipam import V4HostAddress, V6HostAddress


class V4ServiceNetwork(BaseSettings):
    v4: ipaddress.IPv4Network


class V6ServiceNetwork(BaseSettings):
    v6: ipaddress.IPv6Network


class ServiceNetworks(BaseSettings):
    v4: ipaddress.IPv4Network
    v6: ipaddress.IPv6Network


class HostAddresses(BaseSettings):
    v4: ipaddress.IPv4Address
    v6: ipaddress.IPv6Address


def new_service_networks(service_type: str = "", comment: str = "", extattrs: dict = None) -> ServiceNetworks:
    if extattrs is None:
        extattrs = {}
    v4_service_network = _ipam.allocate_service_ipv4_network(
        service_type=service_type, comment=comment, extattrs=extattrs
    )
    v6_service_network = _ipam.allocate_service_ipv6_network(
        service_type=service_type, comment=comment, extattrs=extattrs
    )
    return ServiceNetworks(v4=v4_service_network.v4, v6=v6_service_network.v6)


def new_service_host(
    hostname: str,
    service_type: str = "",
    service_networks: Optional[ServiceNetworks] = None,
    host_addresses: Optional[HostAddresses] = None,
    cname_aliases: list = None,
    extattrs: dict = None,
) -> HostAddresses:
    if extattrs is None:
        extattrs = {}
    return _ipam.allocate_service_host(
        hostname=hostname,
        service_type=service_type,
        service_networks=service_networks,
        host_addresses=host_addresses,
        cname_aliases=cname_aliases,
        extattrs=extattrs,
    )


def delete_service_network(
    network: ipaddress.ip_network = None, service_type: str = ""
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
    return _ipam.delete_service_network(ipnetwork=network, service_type=service_type)


def delete_service_host(
    hostname: str = "", host_addresses: HostAddresses = None, cname_aliases=None, service_type: str = ""
) -> V4HostAddress | V6HostAddress:
    if cname_aliases is None:
        cname_aliases = []
    return _ipam.delete_service_host(
        hostname=hostname, host_addresses=host_addresses, cname_aliases=cname_aliases, service_type=service_type
    )