Source code for galaxy.schema.fetch_data

import json
from enum import Enum
from typing import (
    List,
    Optional,
    Union,
)

from pydantic import (
    BaseModel,
    Extra,
    Field,
    Json,
    validator,
)
from typing_extensions import (
    Annotated,
    Literal,
)

from galaxy.schema.fields import EncodedDatabaseIdField
from .schema import HistoryIdField


[docs]class FetchBaseModel(BaseModel):
[docs] class Config: allow_population_by_field_name = True
[docs]class ElementsFromType(str, Enum): archive = "archive" bagit = "bagit" bagit_archive = "bagit_archive" directory = "directory"
AutoDecompressField = Field(False, description="Decompress compressed data before sniffing?")
[docs]class BaseFetchDataTarget(FetchBaseModel): auto_decompress: bool = AutoDecompressField
[docs]class ItemsFromSrc(str, Enum): url = "url" files = "files" path = "path" ftp_import = "ftp_import" server_dir = "server_dir"
[docs]class Src(str, Enum): url = "url" pasted = "pasted" files = "files" path = "path" composite = "composite" ftp_import = "ftp_import" server_dir = "server_dir"
[docs]class DestinationType(str, Enum): library = "library" library_folder = "library_folder" hdcas = "hdcas" hdas = "hdas"
[docs]class HdaDestination(FetchBaseModel): type: Literal["hdas"]
[docs]class HdcaDestination(FetchBaseModel): type: Literal["hdca"]
[docs]class LibraryFolderDestination(FetchBaseModel): type: Literal["library_folder"] library_folder_id: EncodedDatabaseIdField
[docs]class BaseCollectionTarget(BaseFetchDataTarget): destination: HdcaDestination collection_type: Optional[str] tags: Optional[List[str]] name: Optional[str]
[docs]class LibraryDestination(FetchBaseModel): type: Literal["library"] name: str = Field(..., description="Must specify a library name") description: Optional[str] = Field(None, description="Description for library to create") synopsis: Optional[str] = Field(None, description="Description for library to create")
[docs]class ExtraFiles(FetchBaseModel): items_from: Optional[str] src: Src fuzzy_root: Optional[bool] = Field( True, description="Prevent Galaxy from checking for a single file in a directory and re-interpreting the archive", )
[docs]class BaseDataElement(FetchBaseModel): name: Optional[str] dbkey: str = Field("?") info: Optional[str] ext: str = Field("auto") space_to_tab: bool = False to_posix_lines: bool = False deferred: bool = False tags: Optional[List[str]] created_from_basename: Optional[str] extra_files: Optional[ExtraFiles] auto_decompress: bool = AutoDecompressField items_from: Optional[ElementsFromType] = Field(alias="elements_from") collection_type: Optional[str] MD5: Optional[str]
[docs] class Config: # reject unknown extra attributes extra = Extra.forbid
[docs]class FileDataElement(BaseDataElement): src: Literal["files"]
[docs]class PastedDataElement(BaseDataElement): src: Literal["pasted"] paste_content: str = Field(..., description="Content to upload")
[docs]class UrlDataElement(BaseDataElement): src: Literal["url"] url: str = Field(..., description="URL to upload")
[docs]class ServerDirElement(BaseDataElement): src: Literal["server_dir"] server_dir: str link_data_only: Optional[bool]
[docs]class FtpImportElement(BaseDataElement): src: Literal["ftp_import"] ftp_path: str items_from: Optional[ElementsFromType] = Field(alias="elements_from") name: Optional[str] collection_type: Optional[str]
[docs]class ItemsFromModel(BaseModel): src: ItemsFromSrc path: Optional[str] ftp_path: Optional[str] server_dir: Optional[str] url: Optional[str]
[docs]class FtpImportTarget(BaseCollectionTarget): src: Literal["ftp_import"] ftp_path: str items_from: Optional[ElementsFromType] = Field(alias="elements_from")
[docs]class PathDataElement(BaseDataElement): src: Literal["path"] path: str items_from: Optional[ElementsFromType] = Field(alias="elements_from") link_data_only: Optional[bool]
[docs]class CompositeDataElement(BaseDataElement): src: Literal["composite"] composite: "CompositeItems"
[docs]class CompositeItems(FetchBaseModel): items: List[ Union[FileDataElement, PastedDataElement, UrlDataElement, PathDataElement, ServerDirElement, FtpImportElement] ] = Field(..., alias="elements")
CompositeDataElement.update_forward_refs()
[docs]class NestedElement(BaseDataElement): items: List[Union["AnyElement", "NestedElement"]] = Field(..., alias="elements")
AnyElement = Annotated[ Union[ FileDataElement, PastedDataElement, UrlDataElement, PathDataElement, ServerDirElement, FtpImportElement, CompositeDataElement, ], Field(default_factory=None, discriminator="src"), ] # Seems to be a bug in pydantic ... can't reuse AnyElement in more than one model AnyElement2 = Annotated[ Union[ FileDataElement, PastedDataElement, UrlDataElement, PathDataElement, ServerDirElement, FtpImportElement, CompositeDataElement, ], Field(default_factory=None, discriminator="src"), ] NestedElement.update_forward_refs()
[docs]class BaseDataTarget(BaseFetchDataTarget): destination: Union[HdaDestination, LibraryFolderDestination, LibraryDestination] = Field(..., discriminator="type")
[docs]class DataElementsTarget(BaseDataTarget): items: List[Union[AnyElement, NestedElement]] = Field(..., alias="elements")
[docs]class DataElementsFromTarget(BaseDataTarget, ItemsFromModel): items_from: ElementsFromType = Field(..., alias="elements_from")
[docs]class HdcaDataItemsTarget(BaseCollectionTarget): items: List[Union[AnyElement2, NestedElement]] = Field(..., alias="elements")
[docs]class HdcaDataItemsFromTarget(BaseCollectionTarget, ItemsFromModel): items_from: ElementsFromType = Field(..., alias="elements_from")
[docs]class FilesPayload(BaseModel): filename: str local_filename: str
[docs]class BaseDataPayload(FetchBaseModel): history_id: EncodedDatabaseIdField = HistoryIdField
[docs] class Config: # file payloads are just tacked on, so we need to allow everything extra = Extra.allow
[docs] @validator("targets", pre=True, check_fields=False) def targets_string_to_json(cls, v): if isinstance(v, str): return json.loads(v) return v
Targets = List[ Union[ DataElementsTarget, HdcaDataItemsTarget, DataElementsFromTarget, HdcaDataItemsFromTarget, FtpImportTarget, ] ]
[docs]class FetchDataPayload(BaseDataPayload): targets: Targets
[docs]class FetchDataFormPayload(BaseDataPayload): targets: Union[Json[Targets], Targets] # type: ignore[type-arg] # https://github.com/samuelcolvin/pydantic/issues/2990