Source code for galaxy.schema.fields

import re
from typing import (
    get_origin,
    TYPE_CHECKING,
    Union,
)

from pydantic import (
    BeforeValidator,
    Field,
    PlainSerializer,
    WithJsonSchema,
)
from typing_extensions import (
    Annotated,
    get_args,
)

if TYPE_CHECKING:
    from galaxy.security.idencoding import IdEncodingHelper

ENCODED_DATABASE_ID_PATTERN = re.compile("f?[0-9a-f]+")
ENCODED_ID_LENGTH_MULTIPLE = 16


[docs]class Security:
    security: "IdEncodingHelper"


[docs]def ensure_valid_id(v: str) -> str:
    assert isinstance(v, str)
    len_v = len(v)
    if len_v % ENCODED_ID_LENGTH_MULTIPLE:
        raise ValueError("Invalid id length, must be multiple of 16")
    m = ENCODED_DATABASE_ID_PATTERN.fullmatch(v.lower())
    if not m:
        raise ValueError("Invalid characters in encoded ID")
    return v


[docs]def ensure_valid_folder_id(v):
    if not isinstance(v, str):
        raise ValueError("String required")
    if not v.startswith("F"):
        raise ValueError("Invalid library folder ID. Folder IDs must start with an 'F'")
    v = v[1:]
    ensure_valid_id(v)
    return v


DecodedDatabaseIdField = Annotated[
    int,
    BeforeValidator(lambda database_id: Security.security.decode_id(database_id)),
    BeforeValidator(ensure_valid_id),
    PlainSerializer(lambda database_id: Security.security.encode_id(database_id), return_type=str, when_used="json"),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="serialization",
    ),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="validation",
    ),
]

EncodedDatabaseIdField = Annotated[
    str,
    BeforeValidator(lambda database_id: Security.security.encode_id(database_id)),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="serialization",
    ),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="validation",
    ),
]

LibraryFolderDatabaseIdField = Annotated[
    int,
    BeforeValidator(lambda database_id: Security.security.decode_id(database_id)),
    BeforeValidator(ensure_valid_folder_id),
    PlainSerializer(
        lambda database_id: f"F{Security.security.encode_id(database_id)}", return_type=str, when_used="json"
    ),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="serialization",
    ),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="validation",
    ),
]

EncodedLibraryFolderDatabaseIdField = Annotated[
    str,
    BeforeValidator(lambda database_id: f"F{Security.security.encode_id(database_id)}"),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="serialization",
    ),
    WithJsonSchema(
        {"type": "string", "example": "0123456789ABCDEF", "pattern": "[0-9a-fA-F]+", "minLength": 16},
        mode="validation",
    ),
]


[docs]def literal_to_value(arg):
    val = get_args(arg)
    if not val:
        return arg
    if len(val) > 1:
        raise Exception("Can't extract default argument for unions")
    return val[0]


[docs]def is_optional(field):
    args = get_args(field)
    return get_origin(field) is Union and len(args) == 2 and type(None) in args


[docs]def ModelClassField(default_value):
    """Represents a database model class name annotated as a constant
    pydantic Field.
    :param class_name: The name of the database class.
    :return: A constant pydantic Field with default annotations for model classes.
    """
    return Field(
        ...,
        title="Model class",
        description="The name of the database model class.",
        json_schema_extra={"const": literal_to_value(default_value), "type": "string"},
    )


[docs]def accept_wildcard_defaults_to_json(v):
    assert isinstance(v, str)
    # Accept header can have multiple comma separated values.
    # If any of these values is the wildcard - we default to application/json.
    if "*/*" in v:
        return "application/json"
    return v


AcceptHeaderValidator = BeforeValidator(accept_wildcard_defaults_to_json)