Attachments

albert.collections.attachments.AttachmentCollection

AttachmentCollection(*, session)

Bases: BaseCollection

AttachmentCollection is a collection class for managing Attachment entities in the Albert platform.

Methods:

Name	Description
get_by_id	Retrieves an attachment by its ID.
create	Create a new attachment.
get_by_parent_ids	Retrieves attachments by their parent IDs.
attach_file_to_note	Attaches an already uploaded file to a note.
delete	Deletes an attachment by ID.
upload_and_attach_file_as_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.
upload_and_attach_sds_to_inventory_item	Upload an SDS 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.

Attributes:

Name	Type	Description
base_path

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/{_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())

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.	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.

    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)

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

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.
    """
    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 | None	Jurisdiction code associated with the SDS (e.g. US).	'US'
language_code	str	Language code for the SDS (e.g. EN).	'EN'
hazard_statements	list[HazardStatement] | None	Collection of hazard statements.	None
wgk	str | None	WGK classification metadata.	None

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 | None, optional
        Jurisdiction code associated with the SDS (e.g. ``US``).
    language_code : str, optional
        Language code for the SDS (e.g. ``EN``).
    hazard_statements : list[HazardStatement] | None, optional
        Collection of hazard statements.
    wgk : str | None, optional
        WGK classification metadata.
    """

    sds_path = file_sds.expanduser()
    if not sds_path.is_file():
        raise FileNotFoundError(f"SDS file not found at '{sds_path}'")

    content_type = mimetypes.guess_type(sds_path.name)[0] or "application/pdf"

    extension = sds_path.suffix
    upload_id = self._generate_upload_id()
    encoded_file_name = quote(sds_path.name)
    file_key = f"{inventory_id}/SDS/{upload_id}{extension}"

    file_collection = self._get_file_collection()
    with sds_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] = {
        "jurisdictionCode": jurisdiction_code,
        "languageCode": language_code,
    }

    if revision_date is not None:
        metadata["revisionDate"] = revision_date.isoformat()

    if hazard_statements:
        metadata["hazardStatement"] = [
            statement.model_dump(by_alias=True, exclude_none=True)
            for statement in hazard_statements
        ]
    if hazard_symbols:
        metadata["Symbols"] = [
            symbol.model_dump(by_alias=True, exclude_none=True) for symbol in hazard_symbols
        ]

    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

    attachment = Attachment(
        parent_id=inventory_id,
        name=encoded_file_name,
        key=file_key,
        namespace=FileNamespace.RESULT.value,
        category=AttachmentCategory.SDS,
        metadata=metadata,
        revision_date=revision_date,
    )
    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.

Args: project_id: The Albert ID of the project (e.g. "PRO770"). file_path: Local path to the file to upload.

Returns: 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.

    Args:
        project_id: The Albert ID of the project (e.g. "PRO770").
        file_path: Local path to the file to upload.

    Returns:
        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"
    encoded_file_name = quote(resolved_path.name)
    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=encoded_file_name,
        key=file_key,
        namespace=FileNamespace.RESULT.value,
        category=AttachmentCategory.OTHER,
    )
    return self.create(attachment=attachment)