Class swarmauri_crypto_paramiko.ParamikoCrypto.ParamikoCrypto

swarmauri_crypto_paramiko.ParamikoCrypto.ParamikoCrypto

Bases: CryptoBase

Crypto provider backed by Paramiko and cryptography.

Implements AES-256-GCM for symmetric operations and RSA-OAEP for key wrapping and sealing.

type class-attribute instance-attribute

type = 'ParamikoCrypto'

model_config class-attribute instance-attribute

model_config = ConfigDict(
    extra="allow", arbitrary_types_allowed=True
)

id class-attribute instance-attribute

id = Field(default_factory=generate_id)

members class-attribute instance-attribute

members = None

owners class-attribute instance-attribute

owners = None

host class-attribute instance-attribute

host = None

default_logger class-attribute

default_logger = None

logger class-attribute instance-attribute

logger = None

name class-attribute instance-attribute

name = None

resource class-attribute instance-attribute

resource = Field(default=CRYPTO.value, frozen=True)

version class-attribute instance-attribute

version = '0.1.0'

supports

supports()

Return the algorithms supported by this provider.

RETURNS	DESCRIPTION
Dict[str, Iterable[Alg]]	Dict[str, Iterable[Alg]]: Mapping of operation names to the algorithms the provider can execute.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

def supports(self) -> Dict[str, Iterable[Alg]]:
    """Return the algorithms supported by this provider.

    Returns:
        Dict[str, Iterable[Alg]]: Mapping of operation names to the
            algorithms the provider can execute.
    """

    return {
        "encrypt": (_AEAD_DEFAULT,),
        "decrypt": (_AEAD_DEFAULT,),
        "wrap": (_WRAP_ALG,),
        "unwrap": (_WRAP_ALG,),
        "seal": (_SEAL_ALG,),
        "unseal": (_SEAL_ALG,),
    }

encrypt async

encrypt(key, pt, *, alg=None, aad=None, nonce=None)

Encrypt plaintext with AES-GCM.

PARAMETER	DESCRIPTION
key	Symmetric key reference containing the key bytes. TYPE: KeyRef
pt	Plaintext to encrypt. TYPE: bytes
alg	AEAD algorithm to use. Defaults to AES-256-GCM when not provided. TYPE: Optional[Alg] DEFAULT: None
aad	Additional authenticated data. TYPE: Optional[bytes] DEFAULT: None
nonce	Nonce for AES-GCM. Random when omitted. TYPE: Optional[bytes] DEFAULT: None

RETURNS	DESCRIPTION
AEADCiphertext	Ciphertext container with nonce and tag. TYPE: AEADCiphertext

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported algorithm is requested.
ValueError	If the key material is missing or invalid.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def encrypt(
    self,
    key: KeyRef,
    pt: bytes,
    *,
    alg: Optional[Alg] = None,
    aad: Optional[bytes] = None,
    nonce: Optional[bytes] = None,
) -> AEADCiphertext:
    """Encrypt plaintext with AES-GCM.

    Args:
        key (KeyRef): Symmetric key reference containing the key bytes.
        pt (bytes): Plaintext to encrypt.
        alg (Optional[Alg]): AEAD algorithm to use. Defaults to
            ``AES-256-GCM`` when not provided.
        aad (Optional[bytes]): Additional authenticated data.
        nonce (Optional[bytes]): Nonce for AES-GCM. Random when omitted.

    Returns:
        AEADCiphertext: Ciphertext container with nonce and tag.

    Raises:
        UnsupportedAlgorithm: If an unsupported algorithm is requested.
        ValueError: If the key material is missing or invalid.
    """

    alg = self._normalize_aead_alg(alg)
    if alg != _AEAD_DEFAULT:
        raise UnsupportedAlgorithm(f"Unsupported AEAD algorithm: {alg}")

    if key.material is None:
        raise ValueError(
            "KeyRef.material must contain symmetric key bytes for AEAD"
        )
    if len(key.material) not in (16, 24, 32):
        raise ValueError("KeyRef.material must be 16/24/32 bytes for AES-GCM")

    nonce = nonce or secrets.token_bytes(12)
    aead = AESGCM(key.material)
    ct_with_tag = aead.encrypt(nonce, pt, aad)
    ct, tag = ct_with_tag[:-16], ct_with_tag[-16:]
    return AEADCiphertext(
        kid=key.kid,
        version=key.version,
        alg=alg,
        nonce=nonce,
        ct=ct,
        tag=tag,
        aad=aad,
    )

decrypt async

decrypt(key, ct, *, aad=None)

Decrypt ciphertext produced by :meth:encrypt.

PARAMETER	DESCRIPTION
key	Symmetric key reference containing the key bytes. TYPE: KeyRef
ct	Ciphertext container to decrypt. TYPE: AEADCiphertext
aad	Additional authenticated data. Defaults to the value stored in ct when omitted. TYPE: Optional[bytes] DEFAULT: None

RETURNS	DESCRIPTION
bytes	Decrypted plaintext. TYPE: bytes

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported algorithm is requested.
ValueError	If the key material is missing.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def decrypt(
    self,
    key: KeyRef,
    ct: AEADCiphertext,
    *,
    aad: Optional[bytes] = None,
) -> bytes:
    """Decrypt ciphertext produced by :meth:`encrypt`.

    Args:
        key (KeyRef): Symmetric key reference containing the key bytes.
        ct (AEADCiphertext): Ciphertext container to decrypt.
        aad (Optional[bytes]): Additional authenticated data. Defaults to
            the value stored in ``ct`` when omitted.

    Returns:
        bytes: Decrypted plaintext.

    Raises:
        UnsupportedAlgorithm: If an unsupported algorithm is requested.
        ValueError: If the key material is missing.
    """

    if self._normalize_aead_alg(ct.alg) != _AEAD_DEFAULT:
        raise UnsupportedAlgorithm(f"Unsupported AEAD algorithm: {ct.alg}")
    if key.material is None:
        raise ValueError(
            "KeyRef.material must contain symmetric key bytes for AEAD"
        )

    aead = AESGCM(key.material)
    blob = ct.ct + ct.tag
    return aead.decrypt(ct.nonce, blob, aad or ct.aad)

seal async

seal(recipient, pt, *, alg=_SEAL_ALG)

Seal a small plaintext to a recipient's RSA public key.

PARAMETER	DESCRIPTION
recipient	Recipient key reference with OpenSSH RSA public key bytes in KeyRef.public. TYPE: KeyRef
pt	Plaintext to encrypt. TYPE: bytes
alg	Sealing algorithm identifier. Defaults to RSA-OAEP-SHA256-SEAL. TYPE: Optional[Alg] DEFAULT: _SEAL_ALG

RETURNS	DESCRIPTION
bytes	Sealed ciphertext produced by RSA-OAEP. TYPE: bytes

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported algorithm is requested.
ValueError	If the recipient public key is missing.
IntegrityError	If the plaintext exceeds the RSA-OAEP size limit.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def seal(
    self,
    recipient: KeyRef,
    pt: bytes,
    *,
    alg: Optional[Alg] = _SEAL_ALG,
) -> bytes:
    """Seal a small plaintext to a recipient's RSA public key.

    Args:
        recipient (KeyRef): Recipient key reference with OpenSSH RSA
            public key bytes in ``KeyRef.public``.
        pt (bytes): Plaintext to encrypt.
        alg (Optional[Alg]): Sealing algorithm identifier. Defaults to
            ``RSA-OAEP-SHA256-SEAL``.

    Returns:
        bytes: Sealed ciphertext produced by RSA-OAEP.

    Raises:
        UnsupportedAlgorithm: If an unsupported algorithm is requested.
        ValueError: If the recipient public key is missing.
        IntegrityError: If the plaintext exceeds the RSA-OAEP size limit.
    """

    if alg != _SEAL_ALG:
        raise UnsupportedAlgorithm(f"Unsupported seal alg: {alg}")
    if recipient.public is None:
        raise ValueError("KeyRef.public must contain OpenSSH RSA public key bytes")

    rsa_pub = self._load_rsa_pub_ssh(recipient.public)
    self._seal_size_check(rsa_pub, len(pt))

    sealed = rsa_pub.encrypt(
        pt,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None,
        ),
    )
    return sealed

unseal async

unseal(recipient_priv, sealed, *, alg=_SEAL_ALG)

Unseal ciphertext using the recipient's RSA private key.

PARAMETER	DESCRIPTION
recipient_priv	Recipient key reference containing a PEM-encoded RSA private key in KeyRef.material. TYPE: KeyRef
sealed	Ciphertext produced by :meth:seal. TYPE: bytes
alg	Sealing algorithm identifier. Defaults to RSA-OAEP-SHA256-SEAL. TYPE: Optional[Alg] DEFAULT: _SEAL_ALG

RETURNS	DESCRIPTION
bytes	Decrypted plaintext. TYPE: bytes

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported algorithm is requested.
ValueError	If the private key material is missing.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def unseal(
    self,
    recipient_priv: KeyRef,
    sealed: bytes,
    *,
    alg: Optional[Alg] = _SEAL_ALG,
) -> bytes:
    """Unseal ciphertext using the recipient's RSA private key.

    Args:
        recipient_priv (KeyRef): Recipient key reference containing a
            PEM-encoded RSA private key in ``KeyRef.material``.
        sealed (bytes): Ciphertext produced by :meth:`seal`.
        alg (Optional[Alg]): Sealing algorithm identifier. Defaults to
            ``RSA-OAEP-SHA256-SEAL``.

    Returns:
        bytes: Decrypted plaintext.

    Raises:
        UnsupportedAlgorithm: If an unsupported algorithm is requested.
        ValueError: If the private key material is missing.
    """

    if alg != _SEAL_ALG:
        raise UnsupportedAlgorithm(f"Unsupported seal alg: {alg}")
    if recipient_priv.material is None:
        raise ValueError(
            "KeyRef.material must contain PEM-encoded RSA private key bytes"
        )

    priv = self._load_rsa_priv_pem(recipient_priv.material)
    return priv.decrypt(
        sealed,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None,
        ),
    )

encrypt_for_many async

encrypt_for_many(
    recipients,
    pt,
    *,
    enc_alg=None,
    recipient_wrap_alg=None,
    aad=None,
    nonce=None,
)

Encrypt plaintext for multiple recipients.

Operates in two modes: 1. Sealed: Each recipient receives an individual RSA-OAEP sealed ciphertext when enc_alg is _SEAL_ALG. 2. KEM+AEAD: Generates a shared AES-GCM ciphertext and wraps the session key for each recipient using RSA-OAEP.

PARAMETER	DESCRIPTION
recipients	Sequence of recipients. TYPE: Iterable[KeyRef]
pt	Plaintext to encrypt. TYPE: bytes
enc_alg	Encryption algorithm for the shared ciphertext. Defaults to AES-256-GCM. TYPE: Optional[Alg] DEFAULT: None
recipient_wrap_alg	Wrapping algorithm for the session key. Defaults to RSA-OAEP-SHA256. TYPE: Optional[Alg] DEFAULT: None
aad	Additional authenticated data for AES-GCM. TYPE: Optional[bytes] DEFAULT: None
nonce	Nonce for AES-GCM. Random when omitted. TYPE: Optional[bytes] DEFAULT: None

RETURNS	DESCRIPTION
MultiRecipientEnvelope	Envelope containing ciphertext and per- recipient wrapped keys. TYPE: MultiRecipientEnvelope

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported algorithm is requested.
ValueError	If required key material is missing.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def encrypt_for_many(
    self,
    recipients: Iterable[KeyRef],
    pt: bytes,
    *,
    enc_alg: Optional[Alg] = None,
    recipient_wrap_alg: Optional[Alg] = None,
    aad: Optional[bytes] = None,
    nonce: Optional[bytes] = None,
) -> MultiRecipientEnvelope:
    """Encrypt plaintext for multiple recipients.

    Operates in two modes:
    1. **Sealed**: Each recipient receives an individual RSA-OAEP sealed
       ciphertext when ``enc_alg`` is ``_SEAL_ALG``.
    2. **KEM+AEAD**: Generates a shared AES-GCM ciphertext and wraps the
       session key for each recipient using RSA-OAEP.

    Args:
        recipients (Iterable[KeyRef]): Sequence of recipients.
        pt (bytes): Plaintext to encrypt.
        enc_alg (Optional[Alg]): Encryption algorithm for the shared
            ciphertext. Defaults to ``AES-256-GCM``.
        recipient_wrap_alg (Optional[Alg]): Wrapping algorithm for the
            session key. Defaults to ``RSA-OAEP-SHA256``.
        aad (Optional[bytes]): Additional authenticated data for AES-GCM.
        nonce (Optional[bytes]): Nonce for AES-GCM. Random when omitted.

    Returns:
        MultiRecipientEnvelope: Envelope containing ciphertext and per-
            recipient wrapped keys.

    Raises:
        UnsupportedAlgorithm: If an unsupported algorithm is requested.
        ValueError: If required key material is missing.
    """

    # 1) Sealed-style variant (per-recipient ciphertext; no shared ct)
    if enc_alg == _SEAL_ALG:
        recip_infos: list[RecipientInfo] = []
        for r in recipients:
            if r.public is None:
                raise ValueError(
                    "Recipient KeyRef.public must contain OpenSSH RSA public key bytes"
                )
            rsa_pub = self._load_rsa_pub_ssh(r.public)
            self._seal_size_check(rsa_pub, len(pt))
            sealed = rsa_pub.encrypt(
                pt,
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA256()),
                    algorithm=hashes.SHA256(),
                    label=None,
                ),
            )
            recip_infos.append(
                RecipientInfo(
                    kid=r.kid,
                    version=r.version,
                    wrap_alg=_SEAL_ALG,
                    wrapped_key=sealed,
                    nonce=None,
                )
            )

        # Shared fields empty for sealed variant
        return MultiRecipientEnvelope(
            enc_alg=_SEAL_ALG,
            nonce=b"",
            ct=b"",
            tag=b"",
            recipients=tuple(recip_infos),
            aad=None,  # AAD is not bound in RSA-seal mode
        )

    # 2) Default KEM+AEAD path (shared AES-GCM ct + RSA-wrapped CEK)
    enc_alg = self._normalize_aead_alg(enc_alg)
    if enc_alg != _AEAD_DEFAULT:
        raise UnsupportedAlgorithm(f"Unsupported enc_alg: {enc_alg}")
    wrap_alg = recipient_wrap_alg or _WRAP_ALG
    if wrap_alg != _WRAP_ALG:
        raise UnsupportedAlgorithm(f"Unsupported wrap_alg: {wrap_alg}")

    k = secrets.token_bytes(32)  # 256-bit session key
    iv = nonce or secrets.token_bytes(12)
    aead = AESGCM(k)
    ct_with_tag = aead.encrypt(iv, pt, aad)
    ct, tag = ct_with_tag[:-16], ct_with_tag[-16:]

    recip_infos: list[RecipientInfo] = []
    for r in recipients:
        if r.public is None:
            raise ValueError(
                "Recipient KeyRef.public must contain OpenSSH RSA public key bytes"
            )
        rsa_pub = self._load_rsa_pub_ssh(r.public)
        enc_k = rsa_pub.encrypt(
            k,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None,
            ),
        )
        recip_infos.append(
            RecipientInfo(
                kid=r.kid,
                version=r.version,
                wrap_alg=wrap_alg,
                wrapped_key=enc_k,
            )
        )

    return MultiRecipientEnvelope(
        enc_alg=enc_alg,
        nonce=iv,
        ct=ct,
        tag=tag,
        recipients=tuple(recip_infos),
        aad=aad,
    )

wrap async

wrap(kek, *, dek=None, wrap_alg=None, nonce=None, aad=None)

Wrap a data-encryption key with the given key-encryption key.

PARAMETER	DESCRIPTION
kek	Key-encryption key reference. Provide an OpenSSH RSA public key in KeyRef.public for RSA-OAEP wrapping or symmetric key bytes in KeyRef.material for AES-GCM. TYPE: KeyRef
dek	Data-encryption key to wrap. A random key is generated when omitted. TYPE: Optional[bytes] DEFAULT: None
wrap_alg	Wrapping algorithm to use. Defaults to RSA-OAEP-SHA256. TYPE: Optional[Alg] DEFAULT: None
nonce	Nonce for AES-GCM wrapping. Random when omitted. TYPE: Optional[bytes] DEFAULT: None
aad	Additional authenticated data for AES-GCM wrapping. TYPE: Optional[bytes] DEFAULT: None

RETURNS	DESCRIPTION
WrappedKey	Container with the wrapped key material. TYPE: WrappedKey

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported wrapping algorithm is requested.
ValueError	If required key material is missing or invalid.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def wrap(
    self,
    kek: KeyRef,
    *,
    dek: Optional[bytes] = None,
    wrap_alg: Optional[Alg] = None,
    nonce: Optional[bytes] = None,
    aad: Optional[bytes] = None,
) -> WrappedKey:
    """Wrap a data-encryption key with the given key-encryption key.

    Args:
        kek (KeyRef): Key-encryption key reference. Provide an OpenSSH
            RSA public key in ``KeyRef.public`` for RSA-OAEP wrapping or
            symmetric key bytes in ``KeyRef.material`` for AES-GCM.
        dek (Optional[bytes]): Data-encryption key to wrap. A random key
            is generated when omitted.
        wrap_alg (Optional[Alg]): Wrapping algorithm to use. Defaults to
            ``RSA-OAEP-SHA256``.
        nonce (Optional[bytes]): Nonce for AES-GCM wrapping. Random when
            omitted.
        aad (Optional[bytes]): Additional authenticated data for AES-GCM
            wrapping.

    Returns:
        WrappedKey: Container with the wrapped key material.

    Raises:
        UnsupportedAlgorithm: If an unsupported wrapping algorithm is
            requested.
        ValueError: If required key material is missing or invalid.
    """

    wrap_alg = wrap_alg or _WRAP_ALG
    if wrap_alg == _WRAP_ALG:
        if kek.public is None:
            raise ValueError(
                "KeyRef.public must contain OpenSSH RSA public key bytes"
            )
        rsa_pub = self._load_rsa_pub_ssh(kek.public)
        if dek is None:
            dek = secrets.token_bytes(32)
        wrapped = rsa_pub.encrypt(
            dek,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None,
            ),
        )
        return WrappedKey(
            kek_kid=kek.kid,
            kek_version=kek.version,
            wrap_alg=wrap_alg,
            nonce=nonce,
            wrapped=wrapped,
        )

    alg = self._normalize_aead_alg(wrap_alg)
    if alg != _AEAD_DEFAULT:
        raise UnsupportedAlgorithm(f"Unsupported wrap_alg: {wrap_alg}")
    if kek.material is None:
        raise ValueError(
            "KeyRef.material must contain symmetric key bytes for AES-GCM wrap"
        )
    if len(kek.material) not in (16, 24, 32):
        raise ValueError("KeyRef.material must be 16/24/32 bytes for AES-GCM")
    if dek is None:
        dek = secrets.token_bytes(32)
    nonce = nonce or secrets.token_bytes(12)
    aead = AESGCM(kek.material)
    ct_with_tag = aead.encrypt(nonce, dek, aad)
    ct, tag = ct_with_tag[:-16], ct_with_tag[-16:]
    return WrappedKey(
        kek_kid=kek.kid,
        kek_version=kek.version,
        wrap_alg=wrap_alg,
        nonce=nonce,
        wrapped=ct,
        tag=tag,
    )

unwrap async

unwrap(kek, wrapped, *, aad=None)

Unwrap a previously wrapped key.

PARAMETER	DESCRIPTION
kek	Key-encryption key reference. Must contain the RSA private key or symmetric key bytes corresponding to the wrapping algorithm used. TYPE: KeyRef
wrapped	Wrapped key to unwrap. TYPE: WrappedKey
aad	Additional authenticated data for AES-GCM unwrapping. TYPE: Optional[bytes] DEFAULT: None

RETURNS	DESCRIPTION
bytes	The unwrapped data-encryption key. TYPE: bytes

RAISES	DESCRIPTION
UnsupportedAlgorithm	If an unsupported wrapping algorithm is requested.
ValueError	If key material or required fields are missing.

Source code in swarmauri_crypto_paramiko/ParamikoCrypto.py

async def unwrap(
    self,
    kek: KeyRef,
    wrapped: WrappedKey,
    *,
    aad: Optional[bytes] = None,
) -> bytes:
    """Unwrap a previously wrapped key.

    Args:
        kek (KeyRef): Key-encryption key reference. Must contain the RSA
            private key or symmetric key bytes corresponding to the
            wrapping algorithm used.
        wrapped (WrappedKey): Wrapped key to unwrap.
        aad (Optional[bytes]): Additional authenticated data for AES-GCM
            unwrapping.

    Returns:
        bytes: The unwrapped data-encryption key.

    Raises:
        UnsupportedAlgorithm: If an unsupported wrapping algorithm is
            requested.
        ValueError: If key material or required fields are missing.
    """

    if wrapped.wrap_alg == _WRAP_ALG:
        if kek.material is None:
            raise ValueError(
                "KeyRef.material must contain PEM-encoded RSA private key bytes"
            )
        priv = self._load_rsa_priv_pem(kek.material)
        return priv.decrypt(
            wrapped.wrapped,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None,
            ),
        )

    alg = self._normalize_aead_alg(wrapped.wrap_alg)
    if alg != _AEAD_DEFAULT:
        raise UnsupportedAlgorithm(f"Unsupported wrap_alg: {wrapped.wrap_alg}")
    if kek.material is None:
        raise ValueError(
            "KeyRef.material must contain symmetric key bytes for AES-GCM unwrap"
        )
    if wrapped.nonce is None:
        raise ValueError("WrappedKey.nonce required for AES-GCM unwrap")
    if wrapped.tag is None:
        raise ValueError("WrappedKey.tag required for AES-GCM unwrap")
    aead = AESGCM(kek.material)
    blob = wrapped.wrapped + wrapped.tag
    return aead.decrypt(wrapped.nonce, blob, aad)

register_model classmethod

register_model()

Decorator to register a base model in the unified registry.

RETURNS	DESCRIPTION
Callable	A decorator function that registers the model class. TYPE: Callable[[Type[BaseModel]], Type[BaseModel]]

Source code in swarmauri_base/DynamicBase.py

@classmethod
def register_model(cls) -> Callable[[Type[BaseModel]], Type[BaseModel]]:
    """
    Decorator to register a base model in the unified registry.

    Returns:
        Callable: A decorator function that registers the model class.
    """

    def decorator(model_cls: Type[BaseModel]):
        """Register ``model_cls`` as a base model."""
        model_name = model_cls.__name__
        if model_name in cls._registry:
            glogger.warning(
                "Model '%s' is already registered; skipping duplicate.", model_name
            )
            return model_cls

        cls._registry[model_name] = {"model_cls": model_cls, "subtypes": {}}
        glogger.debug("Registered base model '%s'.", model_name)
        DynamicBase._recreate_models()
        return model_cls

    return decorator

register_type classmethod

register_type(resource_type=None, type_name=None)

Decorator to register a subtype under one or more base models in the unified registry.

PARAMETER	DESCRIPTION
resource_type	The base model(s) under which to register the subtype. If None, all direct base classes (except DynamicBase) are used. TYPE: Optional[Union[Type[T], List[Type[T]]]] DEFAULT: None
type_name	An optional custom type name for the subtype. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Callable	A decorator function that registers the subtype. TYPE: Callable[[Type[DynamicBase]], Type[DynamicBase]]

Source code in swarmauri_base/DynamicBase.py

@classmethod
def register_type(
    cls,
    resource_type: Optional[Union[Type[T], List[Type[T]]]] = None,
    type_name: Optional[str] = None,
) -> Callable[[Type["DynamicBase"]], Type["DynamicBase"]]:
    """
    Decorator to register a subtype under one or more base models in the unified registry.

    Parameters:
        resource_type (Optional[Union[Type[T], List[Type[T]]]]):
            The base model(s) under which to register the subtype. If None, all direct base classes (except DynamicBase)
            are used.
        type_name (Optional[str]): An optional custom type name for the subtype.

    Returns:
        Callable: A decorator function that registers the subtype.
    """

    def decorator(subclass: Type["DynamicBase"]):
        """Register ``subclass`` as a subtype."""
        if resource_type is None:
            resource_types = [
                base for base in subclass.__bases__ if base is not cls
            ]
        elif not isinstance(resource_type, list):
            resource_types = [resource_type]
        else:
            resource_types = resource_type

        for rt in resource_types:
            if not issubclass(subclass, rt):
                raise TypeError(
                    f"'{subclass.__name__}' must be a subclass of '{rt.__name__}'."
                )
            final_type_name = type_name or getattr(
                subclass, "_type", subclass.__name__
            )
            base_model_name = rt.__name__

            if base_model_name not in cls._registry:
                cls._registry[base_model_name] = {"model_cls": rt, "subtypes": {}}
                glogger.debug(
                    "Created new registry entry for base model '%s'.",
                    base_model_name,
                )

            subtypes_dict = cls._registry[base_model_name]["subtypes"]
            if final_type_name in subtypes_dict:
                glogger.warning(
                    "Type '%s' already exists under '%s'; skipping duplicate.",
                    final_type_name,
                    base_model_name,
                )
                continue

            subtypes_dict[final_type_name] = subclass
            glogger.debug(
                "Registered '%s' as '%s' under '%s'.",
                subclass.__name__,
                final_type_name,
                base_model_name,
            )

        DynamicBase._recreate_models()
        return subclass

    return decorator

model_validate_toml classmethod

model_validate_toml(toml_data)

Validate a model from a TOML string.

Source code in swarmauri_base/TomlMixin.py

@classmethod
def model_validate_toml(cls, toml_data: str):
    """Validate a model from a TOML string."""
    try:
        # Parse TOML into a Python dictionary
        toml_content = tomllib.loads(toml_data)

        # Convert the dictionary to JSON and validate using Pydantic
        return cls.model_validate_json(json.dumps(toml_content))
    except tomllib.TOMLDecodeError as e:
        raise ValueError(f"Invalid TOML data: {e}")
    except ValidationError as e:
        raise ValueError(f"Validation failed: {e}")

model_dump_toml

model_dump_toml(
    fields_to_exclude=None, api_key_placeholder=None
)

Return a TOML representation of the model.

Source code in swarmauri_base/TomlMixin.py

def model_dump_toml(self, fields_to_exclude=None, api_key_placeholder=None):
    """Return a TOML representation of the model."""
    if fields_to_exclude is None:
        fields_to_exclude = []

    # Load the JSON string into a Python dictionary
    json_data = json.loads(self.model_dump_json())

    # Function to recursively remove specific keys and handle api_key placeholders
    def process_fields(data, fields_to_exclude):
        """Recursively filter fields and apply placeholders."""
        if isinstance(data, dict):
            return {
                key: (
                    api_key_placeholder
                    if key == "api_key" and api_key_placeholder is not None
                    else process_fields(value, fields_to_exclude)
                )
                for key, value in data.items()
                if key not in fields_to_exclude
            }
        elif isinstance(data, list):
            return [process_fields(item, fields_to_exclude) for item in data]
        else:
            return data

    # Filter the JSON data
    filtered_data = process_fields(json_data, fields_to_exclude)

    # Convert the filtered data into TOML
    return toml.dumps(filtered_data)

model_validate_yaml classmethod

model_validate_yaml(yaml_data)

Validate a model from a YAML string.

Source code in swarmauri_base/YamlMixin.py

@classmethod
def model_validate_yaml(cls, yaml_data: str):
    """Validate a model from a YAML string."""
    try:
        # Parse YAML into a Python dictionary
        yaml_content = yaml.safe_load(yaml_data)

        # Convert the dictionary to JSON and validate using Pydantic
        return cls.model_validate_json(json.dumps(yaml_content))
    except yaml.YAMLError as e:
        raise ValueError(f"Invalid YAML data: {e}")
    except ValidationError as e:
        raise ValueError(f"Validation failed: {e}")

model_dump_yaml

model_dump_yaml(
    fields_to_exclude=None, api_key_placeholder=None
)

Return a YAML representation of the model.

Source code in swarmauri_base/YamlMixin.py

def model_dump_yaml(self, fields_to_exclude=None, api_key_placeholder=None):
    """Return a YAML representation of the model."""
    if fields_to_exclude is None:
        fields_to_exclude = []

    # Load the JSON string into a Python dictionary
    json_data = json.loads(self.model_dump_json())

    # Function to recursively remove specific keys and handle api_key placeholders
    def process_fields(data, fields_to_exclude):
        """Recursively filter fields and apply placeholders."""
        if isinstance(data, dict):
            return {
                key: (
                    api_key_placeholder
                    if key == "api_key" and api_key_placeholder is not None
                    else process_fields(value, fields_to_exclude)
                )
                for key, value in data.items()
                if key not in fields_to_exclude
            }
        elif isinstance(data, list):
            return [process_fields(item, fields_to_exclude) for item in data]
        else:
            return data

    # Filter the JSON data
    filtered_data = process_fields(json_data, fields_to_exclude)

    # Convert the filtered data into YAML using safe mode
    return yaml.safe_dump(filtered_data, default_flow_style=False)

model_post_init

model_post_init(logger=None)

Assign a logger instance after model initialization.

Source code in swarmauri_base/LoggerMixin.py

def model_post_init(self, logger: Optional[FullUnion[LoggerBase]] = None) -> None:
    """Assign a logger instance after model initialization."""

    # Directly assign the provided FullUnion[LoggerBase] or fallback to the
    # class-level default.
    self.logger = self.logger or logger or self.default_logger