Attachments

albert.collections.attachments.AttachmentCollection

AttachmentCollection(*, session)

Bases: BaseCollection

AttachmentCollection manages Attachment entities in the Albert platform.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert session instance.	required

Attributes:

Name	Type	Description
base_path	str	The base URL for attachment API requests.

Methods:

Name	Description
get_by_id	Retrieve an attachment by its ID.
create	Create a new attachment.
delete	Delete an attachment by its ID.
update	Update an existing attachment.
get_by_parent_ids	Retrieve attachments grouped by parent ID.
get_jurisdiction_codes	List available jurisdiction codes.
get_language_codes	List available language codes.
attach_file_to_note	Attach an already-uploaded file to a note.
upload_and_attach_file_to_note	Upload a file and attach it to a note.
upload_and_attach_sds_to_inventory_item	Upload an SDS document and attach it to an inventory item.
upload_and_attach_document_to_inventory_item	Upload a document and attach it to an inventory item.
upload_and_attach_document_to_project	Upload a file and attach it as a document to a project.

Source code in src/albert/collections/attachments.py

def __init__(self, *, session):
    super().__init__(session=session)
    self.base_path = f"/api/{AttachmentCollection._api_version}/attachments"

base_path

base_path = (
    f"/api/{AttachmentCollection._api_version}/attachments"
)

get_by_id

get_by_id(*, id: AttachmentId) -> Attachment

Retrieves an attachment by its ID.

Parameters:

Name	Type	Description	Default
id	AttachmentId	The ID of the attachment to retrieve.	required

Returns:

Type	Description
Attachment	The Attachment object corresponding to the provided ID.

Source code in src/albert/collections/attachments.py

@validate_call
def get_by_id(self, *, id: AttachmentId) -> Attachment:
    """Retrieves an attachment by its ID.

    Parameters
    ----------
    id : AttachmentId
        The ID of the attachment to retrieve.

    Returns
    -------
    Attachment
        The Attachment object corresponding to the provided ID.
    """
    response = self.session.get(url=f"{self.base_path}/{id}")
    return Attachment(**response.json())

create

create(*, attachment: Attachment) -> Attachment

Create a new attachment.

Parameters:

Name	Type	Description	Default
attachment	Attachment	The attachment to create.	required

Returns:

Type	Description
Attachment	The created attachment.

Source code in src/albert/collections/attachments.py

@validate_call
def create(self, *, attachment: Attachment) -> Attachment:
    """Create a new attachment.

    Parameters
    ----------
    attachment : Attachment
        The attachment to create.

    Returns
    -------
    Attachment
        The created attachment.
    """
    payload = attachment.model_dump(by_alias=True, exclude_unset=True, mode="json")
    response = self.session.post(self.base_path, json=payload)
    return Attachment(**response.json())

update

update(*, attachment: Attachment) -> Attachment

Update an attachment.

Parameters:

Name	Type	Description	Default
attachment	Attachment	The attachment with updated fields. Must include id.	required

Returns:

Type	Description
Attachment	The updated Attachment.

Notes

The following fields can be updated: name, parent_id, revision_date.

Source code in src/albert/collections/attachments.py

@validate_call
def update(self, *, attachment: Attachment) -> Attachment:
    """Update an attachment.

    Parameters
    ----------
    attachment : Attachment
        The attachment with updated fields. Must include ``id``.

    Returns
    -------
    Attachment
        The updated Attachment.

    Notes
    -----
    The following fields can be updated: ``name``, ``parent_id``, ``revision_date``.
    """
    if attachment.id is None:
        raise ValueError("Attachment ID is required for update.")

    existing_attachment = self.get_by_id(id=attachment.id)
    payload = self._generate_attachment_patch_payload(
        existing=existing_attachment, updated=attachment
    )
    if len(payload.data) == 0:
        return existing_attachment

    # The API rejects more than one operation on the same list attribute
    # (e.g. Symbols) in a single request, so each such op is sent separately.
    list_attrs = {
        d.attribute
        for d in payload.data
        if isinstance(d.old_value, list) or isinstance(d.new_value, list)
    }
    scalar_data = [d for d in payload.data if d.attribute not in list_attrs]
    list_data = [d for d in payload.data if d.attribute in list_attrs]
    batches = [scalar_data] if scalar_data else []
    batches.extend([d] for d in list_data)
    for batch in batches:
        self.session.patch(
            f"{self.base_path}/{attachment.id}",
            json=PatchPayload(data=batch).model_dump(by_alias=True, mode="json"),
        )
    return self.get_by_id(id=attachment.id)

get_by_parent_ids

get_by_parent_ids(
    *,
    parent_ids: list[str],
    data_column_ids: list[DataColumnId] | None = None,
) -> dict[str, list[Attachment]]

Retrieves attachments by their parent IDs.

Note: This method returns a dictionary where the keys are parent IDs and the values are lists of Attachment objects associated with each parent ID. If the parent ID has no attachments, it will not be included in the dictionary.

If no attachments are found for any of the provided parent IDs, the API response will be an error.

Parameters:

Name	Type	Description	Default
parent_ids	list[str]	Parent IDs of the objects to which the attachments are linked. IDs must include the full entity prefix (e.g. "INVA123" for an inventory item, "PRO123" for a project).	required

Returns:

Type	Description
dict[str, list[Attachment]]	A dictionary mapping parent IDs to lists of Attachment objects associated with each parent ID.

Source code in src/albert/collections/attachments.py

def get_by_parent_ids(
    self, *, parent_ids: list[str], data_column_ids: list[DataColumnId] | None = None
) -> dict[str, list[Attachment]]:
    """Retrieves attachments by their parent IDs.

    Note: This method returns a dictionary where the keys are parent IDs
    and the values are lists of Attachment objects associated with each parent ID.
    If the parent ID has no attachments, it will not be included in the dictionary.

    If no attachments are found for any of the provided parent IDs,
    the API response will be an error.

    Parameters
    ----------
    parent_ids : list[str]
        Parent IDs of the objects to which the attachments are linked. IDs must
        include the full entity prefix (e.g. ``"INVA123"`` for an inventory item,
        ``"PRO123"`` for a project).

    Returns
    -------
    dict[str, list[Attachment]]
        A dictionary mapping parent IDs to lists of Attachment objects associated with each parent ID.
    """
    response = self.session.get(
        url=f"{self.base_path}/parents",
        params={"id": parent_ids, "dataColumnId": data_column_ids},
    )
    response_data = response.json()
    return {
        parent["parentId"]: [
            Attachment(**item, parent_id=parent["parentId"]) for item in parent["Items"]
        ]
        for parent in response_data
    }

attach_file_to_note

attach_file_to_note(
    *,
    note_id: str,
    file_name: str,
    file_key: str,
    category: FileCategory = OTHER,
) -> Attachment

Attaches an already uploaded file to a note.

Parameters:

Name	Type	Description	Default
note_id	str	The ID of the note to attach the file to.	required
file_name	str	The name of the file to attach.	required
file_key	str	The unique key of the file to attach (the returned upload name).	required
category	FileCategory	The type of file, by default FileCategory.OTHER	OTHER

Returns:

Type	Description
Attachment	The related attachment object.

Source code in src/albert/collections/attachments.py

def attach_file_to_note(
    self,
    *,
    note_id: str,
    file_name: str,
    file_key: str,
    category: FileCategory = FileCategory.OTHER,
) -> Attachment:
    """Attaches an already uploaded file to a note.

    Parameters
    ----------
    note_id : str
        The ID of the note to attach the file to.
    file_name : str
        The name of the file to attach.
    file_key : str
        The unique key of the file to attach (the returned upload name).
    category : FileCategory, optional
        The type of file, by default FileCategory.OTHER

    Returns
    -------
    Attachment
        The related attachment object.
    """
    attachment = Attachment(
        parent_id=note_id, name=file_name, key=file_key, namespace="result", category=category
    )
    return self.create(attachment=attachment)

get_jurisdiction_codes

get_jurisdiction_codes() -> dict[str, str]

Return available jurisdiction codes.

Returns:

Type	Description
dict[str, str]	Mapping of jurisdiction name to code (e.g. {"Germany": "DE", "USA": "US"}).

Source code in src/albert/collections/attachments.py

def get_jurisdiction_codes(self) -> dict[str, str]:
    """Return available jurisdiction codes.

    Returns
    -------
    dict[str, str]
        Mapping of jurisdiction name to code (e.g. ``{"Germany": "DE", "USA": "US"}``).
    """
    response = self.session.get(
        f"{self.base_path}/jurisdictionslanguages", params={"type": "jurisdiction"}
    )
    return response.json()

get_language_codes

get_language_codes() -> dict[str, str]

Return available language codes.

Returns:

Type	Description
dict[str, str]	Mapping of language name to code (e.g. {"English": "EN", "German": "DE"}).

Source code in src/albert/collections/attachments.py

def get_language_codes(self) -> dict[str, str]:
    """Return available language codes.

    Returns
    -------
    dict[str, str]
        Mapping of language name to code (e.g. ``{"English": "EN", "German": "DE"}``).
    """
    response = self.session.get(
        f"{self.base_path}/jurisdictionslanguages", params={"type": "language"}
    )
    return response.json()

delete

delete(*, id: AttachmentId) -> None

Deletes an attachment by ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the attachment to delete.	required

Returns:

Type	Description
None

Source code in src/albert/collections/attachments.py

@validate_call
def delete(self, *, id: AttachmentId) -> None:
    """Deletes an attachment by ID.

    Parameters
    ----------
    id : str
        The ID of the attachment to delete.

    Returns
    -------
    None
    """
    self.session.delete(f"{self.base_path}/{id}")

upload_and_attach_file_as_note

upload_and_attach_file_as_note(
    parent_id: str,
    file_data: IO,
    note_text: str = "",
    file_name: str = "",
    upload_key: str | None = None,
) -> Note

Uploads a file and attaches it to a new note. A user can be tagged in the note_text string by using f-string and the User.to_note_mention() method. This allows for easy tagging and referencing of users within notes. example: f"Hello {tagged_user.to_note_mention()}!"

Parameters:

Name	Type	Description	Default
parent_id	str	The ID of the parent entity onto which the note will be attached.	required
file_data	IO	The file data to upload.	required
note_text	str	Any additional text to add to the note, by default ""	''
file_name	str	The name of the file. Include a file extension to infer the content type; otherwise, the upload defaults to application/octet-stream.	''
upload_key	str | None	Override the storage key used when signing and uploading the file. Defaults to {parent_id}/{note_id}/{file_name}.	None

Returns:

Type	Description
Note	The created note.

Source code in src/albert/collections/attachments.py

def upload_and_attach_file_as_note(
    self,
    parent_id: str,
    file_data: IO,
    note_text: str = "",
    file_name: str = "",
    upload_key: str | None = None,
) -> Note:
    """Uploads a file and attaches it to a new note. A user can be tagged in the note_text string by using f-string and the User.to_note_mention() method.
    This allows for easy tagging and referencing of users within notes. example: f"Hello {tagged_user.to_note_mention()}!"

    Parameters
    ----------
    parent_id : str
        The ID of the parent entity onto which the note will be attached.
    file_data : IO
        The file data to upload.
    note_text : str, optional
        Any additional text to add to the note, by default ""
    file_name : str, optional
        The name of the file. Include a file extension to infer the content type;
        otherwise, the upload defaults to ``application/octet-stream``.
    upload_key : str | None, optional
        Override the storage key used when signing and uploading the file.
        Defaults to ``{parent_id}/{note_id}/{file_name}``.

    Returns
    -------
    Note
        The created note.
    """
    if not (upload_key or file_name):
        raise ValueError("A file name or upload key must be provided for attachment upload.")

    note_collection = self._get_note_collection()
    note = Note(
        parent_id=parent_id,
        note=note_text,
    )
    registered_note = note_collection.create(note=note)
    if upload_key:
        attachment_name = file_name or Path(upload_key).name
        upload_name = upload_key
    else:
        attachment_name = file_name
        upload_name = f"{parent_id}/{registered_note.id}/{file_name}"
    file_type = mimetypes.guess_type(attachment_name or upload_name)[0]
    if file_type is None:
        file_type = "application/octet-stream"
    file_collection = self._get_file_collection()

    file_collection.sign_and_upload_file(
        data=file_data,
        name=upload_name,
        namespace=FileNamespace.RESULT.value,
        content_type=file_type,
    )
    file_info = file_collection.get_by_name(
        name=upload_name, namespace=FileNamespace.RESULT.value
    )
    self.attach_file_to_note(
        note_id=registered_note.id,
        file_name=attachment_name,
        file_key=file_info.name,
    )

    return note_collection.get_by_id(id=registered_note.id)

upload_and_attach_sds_to_inventory_item

upload_and_attach_sds_to_inventory_item(
    *,
    inventory_id: InventoryId,
    file_sds: Path,
    revision_date: date,
    storage_class: str,
    un_number: str,
    jurisdiction_code: str = "US",
    language_code: str = "EN",
    hazard_statements: list[HazardStatement] | None = None,
    hazard_symbols: list[HazardSymbol] | None = None,
    wgk: str | None = None,
) -> Attachment

Upload an SDS document and attach it to an inventory item.

Parameters:

Name	Type	Description	Default
inventory_id	str	Id of Inventory Item to attach SDS to.	required
file_sds	Path	Local path to the SDS PDF to upload.	required
revision_date	date	Revision date for the SDS. (yyyy-mm-dd)	required
un_number	str	The UN number.	required
storage_class	str	The Storage Class number.	required
jurisdiction_code	str	Jurisdiction code for the SDS (e.g. "US"). Use get_jurisdiction_codes() to retrieve the full list of available codes.	'US'
language_code	str	Language code for the SDS (e.g. "EN"). Use get_language_codes() to retrieve the full list of available codes.	'EN'
hazard_statements	list[HazardStatement] | None	Collection of hazard statements.	None
hazard_symbols	list[HazardSymbol] | None	Collection of hazard symbols.	None
wgk	str | None	WGK classification metadata.	None

Returns:

Type	Description
Attachment	The created SDS attachment linked to the inventory item.

Source code in src/albert/collections/attachments.py

@validate_call
def upload_and_attach_sds_to_inventory_item(
    self,
    *,
    inventory_id: InventoryId,
    file_sds: Path,
    revision_date: date,
    storage_class: str,
    un_number: str,
    jurisdiction_code: str = "US",
    language_code: str = "EN",
    hazard_statements: list[HazardStatement] | None = None,
    hazard_symbols: list[HazardSymbol] | None = None,
    wgk: str | None = None,
) -> Attachment:
    """Upload an SDS document and attach it to an inventory item.

    Parameters
    ----------
    inventory_id : str
        Id of Inventory Item to attach SDS to.
    file_sds : Path
        Local path to the SDS PDF to upload.
    revision_date : date
        Revision date for the SDS. (yyyy-mm-dd)
    un_number : str
        The UN number.
    storage_class : str
        The Storage Class number.
    jurisdiction_code : str, optional
        Jurisdiction code for the SDS (e.g. ``"US"``). Use
        ``get_jurisdiction_codes()`` to retrieve the full list of available codes.
    language_code : str, optional
        Language code for the SDS (e.g. ``"EN"``). Use
        ``get_language_codes()`` to retrieve the full list of available codes.
    hazard_statements : list[HazardStatement] | None, optional
        Collection of hazard statements.
    hazard_symbols : list[HazardSymbol] | None, optional
        Collection of hazard symbols.
    wgk : str | None, optional
        WGK classification metadata.

    Returns
    -------
    Attachment
        The created SDS attachment linked to the inventory item.
    """

    return self.upload_and_attach_document_to_inventory_item(
        inventory_id=inventory_id,
        file_path=file_sds,
        category=AttachmentCategory.SDS,
        revision_date=revision_date,
        jurisdiction_code=jurisdiction_code,
        language_code=language_code,
        hazard_statements=hazard_statements,
        hazard_symbols=hazard_symbols,
        un_number=un_number,
        storage_class=storage_class,
        wgk=wgk,
    )

upload_and_attach_document_to_inventory_item

upload_and_attach_document_to_inventory_item(
    *,
    inventory_id: InventoryId,
    file_path: Path,
    category: AttachmentCategory | str,
    revision_date: date | None = None,
    description: str | None = None,
    jurisdiction_code: str | None = None,
    language_code: str | None = None,
    hazard_statements: list[HazardStatement] | None = None,
    hazard_symbols: list[HazardSymbol] | None = None,
    un_number: str | None = None,
    storage_class: str | None = None,
    wgk: str | None = None,
) -> Attachment

Upload a file and attach it as a document to an inventory item.

Parameters:

Name	Type	Description	Default
inventory_id	str	The Albert ID of the inventory item (e.g. INVA123).	required
file_path	Path	Local path to the file to upload.	required
category	AttachmentCategory | str	Document category.	required
revision_date	date | None	Revision date for the document.	None
description	str | None	Description for the document.	None
jurisdiction_code	str | None	Jurisdiction code (e.g. "US").	None
language_code	str | None	Language code (e.g. "EN").	None
hazard_statements	list[HazardStatement] | None	Hazard statements to associate with the document.	None
hazard_symbols	list[HazardSymbol] | None	Hazard symbols to associate with the document.	None
un_number	str | None	UN number.	None
storage_class	str | None	Storage class.	None
wgk	str | None	Water hazard class (WGK).	None

Returns:

Type	Description
Attachment	The created attachment linked to the inventory item.

Raises:

Type	Description
FileNotFoundError	If file_path does not exist.

Source code in src/albert/collections/attachments.py

@validate_call
def upload_and_attach_document_to_inventory_item(
    self,
    *,
    inventory_id: InventoryId,
    file_path: Path,
    category: AttachmentCategory | str,
    revision_date: date | None = None,
    description: str | None = None,
    jurisdiction_code: str | None = None,
    language_code: str | None = None,
    hazard_statements: list[HazardStatement] | None = None,
    hazard_symbols: list[HazardSymbol] | None = None,
    un_number: str | None = None,
    storage_class: str | None = None,
    wgk: str | None = None,
) -> Attachment:
    """Upload a file and attach it as a document to an inventory item.

    Parameters
    ----------
    inventory_id : str
        The Albert ID of the inventory item (e.g. ``INVA123``).
    file_path : Path
        Local path to the file to upload.
    category : AttachmentCategory | str
        Document category.
    revision_date : date | None, optional
        Revision date for the document.
    description : str | None, optional
        Description for the document.
    jurisdiction_code : str | None, optional
        Jurisdiction code (e.g. ``"US"``).
    language_code : str | None, optional
        Language code (e.g. ``"EN"``).
    hazard_statements : list[HazardStatement] | None, optional
        Hazard statements to associate with the document.
    hazard_symbols : list[HazardSymbol] | None, optional
        Hazard symbols to associate with the document.
    un_number : str | None, optional
        UN number.
    storage_class : str | None, optional
        Storage class.
    wgk : str | None, optional
        Water hazard class (WGK).

    Returns
    -------
    Attachment
        The created attachment linked to the inventory item.

    Raises
    ------
    FileNotFoundError
        If ``file_path`` does not exist.
    """
    resolved_path = file_path.expanduser()
    if not resolved_path.is_file():
        raise FileNotFoundError(f"File not found at '{resolved_path}'")

    content_type = mimetypes.guess_type(resolved_path.name)[0] or "application/pdf"
    upload_id = self._generate_upload_id()
    file_key = f"{inventory_id}/{FileCategory.SDS.value}/{upload_id}{resolved_path.name}"

    file_collection = self._get_file_collection()
    with resolved_path.open("rb") as file_handle:
        file_collection.sign_and_upload_file(
            data=file_handle,
            name=file_key,
            namespace=FileNamespace.RESULT,
            content_type=content_type,
            category=FileCategory.SDS,
        )

    metadata: dict[str, MetadataItem] = {}
    if revision_date is not None:
        metadata["revisionDate"] = revision_date.isoformat()
    if description is not None:
        metadata["description"] = description
    if jurisdiction_code is not None:
        metadata["jurisdictionCode"] = jurisdiction_code
    if language_code is not None:
        metadata["languageCode"] = language_code
    if un_number is not None:
        metadata["unNumber"] = un_number
    if storage_class is not None:
        metadata["storageClass"] = storage_class
    if wgk is not None:
        metadata["wgk"] = wgk
    if hazard_statements:
        metadata["hazardStatement"] = [
            s.model_dump(by_alias=True, exclude_none=True) for s in hazard_statements
        ]
    if hazard_symbols:
        metadata["Symbols"] = [
            s.model_dump(by_alias=True, exclude_none=True) for s in hazard_symbols
        ]

    attachment_kwargs: dict = {
        "parent_id": inventory_id,
        "name": resolved_path.name,
        "key": file_key,
        "namespace": FileNamespace.RESULT.value,
        "category": category,
    }
    if revision_date is not None:
        attachment_kwargs["revision_date"] = revision_date
    if metadata:
        attachment_kwargs["metadata"] = metadata

    attachment = Attachment(**attachment_kwargs)
    return self.create(attachment=attachment)

upload_and_attach_document_to_project

upload_and_attach_document_to_project(
    *, project_id: ProjectId, file_path: Path
) -> Attachment

Upload a file and attach it as a document to a project.

Parameters:

Name	Type	Description	Default
project_id	ProjectId	The Albert ID of the project (e.g. PRO770).	required
file_path	Path	Local path to the file to upload.	required

Returns:

Type	Description
Attachment	The created attachment record.

Source code in src/albert/collections/attachments.py

@validate_call
def upload_and_attach_document_to_project(
    self,
    *,
    project_id: ProjectId,
    file_path: Path,
) -> Attachment:
    """Upload a file and attach it as a document to a project.

    Parameters
    ----------
    project_id : ProjectId
        The Albert ID of the project (e.g. ``PRO770``).
    file_path : Path
        Local path to the file to upload.

    Returns
    -------
    Attachment
        The created attachment record.
    """
    resolved_path = file_path.expanduser()
    if not resolved_path.is_file():
        raise FileNotFoundError(f"File not found at '{resolved_path}'")

    content_type = mimetypes.guess_type(resolved_path.name)[0] or "application/octet-stream"
    upload_id = self._generate_upload_id()
    extension = resolved_path.suffix
    file_key = f"{project_id}/documents/original/{upload_id}{extension}"

    file_collection = self._get_file_collection()
    with resolved_path.open("rb") as file_handle:
        file_collection.sign_and_upload_file(
            data=file_handle,
            name=file_key,
            namespace=FileNamespace.RESULT,
            content_type=content_type,
        )

    attachment = Attachment(
        parent_id=project_id,
        name=resolved_path.name,
        key=file_key,
        namespace=FileNamespace.RESULT.value,
        category=AttachmentCategory.OTHER,
    )
    return self.create(attachment=attachment)