sqlalchemy_library ¤

def add_collection(
    self,
    name: str,
    user_id: Optional[Union[str, uuid.UUID]] = None,
    track_ids: Optional[List[Union[str, uuid.UUID]]] = None,
    description: str = "",
    collection_type: str = "collection",
    visibility: schema.Visibility = schema.Visibility.private,
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoCollection:
    """Creates a new collection and saves it into the DB.

    Args:
        track_ids (List[Union[str, uuid.UUID]]): List of track ids
            to be added to the collection.
        name (str): Name of the collection.
        user_id (UUID, optional): The ID of the user adding the collection.
        description (str): Description of the collection.
        collection_type (str): Type of the collection.
        meta (Dict[str, Any]): Metadata of the collection.

    Returns:
        NendoCollection: The newly created NendoCollection object.
    """
    user_id = self._ensure_user_uuid(user_id)
    if track_ids is None:
        track_ids = []
    with self.session_scope() as session:
        # Fetch the track objects
        track_objs = session.query(model.NendoTrackDB).filter(
            model.NendoTrackDB.id.in_(
                [uuid.UUID(t) if isinstance(t, str) else t for t in track_ids],
            ),
        )
        if user_id is not None:
            track_objs = track_objs.filter(model.NendoTrackDB.user_id == user_id)
        track_objs = track_objs.all()
        # Create a new collection object
        new_collection = model.NendoCollectionDB(
            name=name,
            user_id=user_id,
            description=description,
            collection_type=collection_type,
            visibility=visibility,
            meta=meta or {},
        )
        session.add(new_collection)
        session.commit()
        session.refresh(new_collection)
        # Create relationships from tracks to collection
        for idx, track in enumerate(track_objs):
            tc_relationship = model.TrackCollectionRelationshipDB(
                source_id=track.id,
                target_id=new_collection.id,
                relationship_type="track",
                meta={},
                relationship_position=idx,
            )
            session.add(tc_relationship)
        session.commit()
        session.refresh(new_collection)

        return schema.NendoCollection.model_validate(new_collection)

def add_plugin_data(
    self,
    track_id: Union[str, uuid.UUID],
    key: str,
    value: Any,
    plugin_name: str,
    plugin_version: Optional[str] = None,
    user_id: Optional[Union[str, uuid.UUID]] = None,
    replace: Optional[bool] = None,
) -> schema.NendoPluginData:
    """Add plugin data to a NendoTrack and persist changes into the DB.

    Args:
        track_id (Union[str, UUID]): ID of the track to which
            the plugin data should be added.
        key (str): Key under which to save the data.
        value (Any): Data to save.
        plugin_name (str): Name of the plugin.
        plugin_version (str, optional): Version of the plugin. If none is given,
            the currently loaded version of the plugin given by `plugin_name`
            will be used.
        user_id (Union[str, UUID], optional): ID of user adding the plugin data.
        replace (bool, optional): Flag that determines whether
            the last existing data point for the given plugin name and -version
            is overwritten or not. If undefined, the nendo configuration's
            `replace_plugin_data` value will be used.

    Returns:
        NendoPluginData: The saved plugin data as a NendoPluginData object.
    """
    # create plugin data
    user_id = self._ensure_user_uuid(user_id)
    replace = (
        replace
        if replace is not None
        else self.nendo_instance.config.replace_plugin_data
    )
    value_converted = self._convert_plugin_data(value=value, user_id=user_id)
    if plugin_version is None:
        try:
            plugin = getattr(self.nendo_instance.plugins, plugin_name)
        except AttributeError as e:  # noqa: F841
            self.logger.error(
                f"Plugin with name {plugin_name} is not loaded. "
                "You have to manually specify the plugin_version parameter.",
            )
            return None
        plugin_version = plugin.version
    plugin_data = schema.NendoPluginDataCreate(
        track_id=ensure_uuid(track_id),
        user_id=user_id,
        plugin_name=plugin_name,
        plugin_version=plugin_version,
        key=key,
        value=value_converted,
    )
    with self.session_scope() as session:
        existing_plugin_data = self._get_latest_plugin_data_db(
            track_id=track_id,
            plugin_name=plugin_name,
            plugin_version=plugin_version,
            key=key,
            session=session,
            user_id=user_id,
        )
        if replace is True:
            if existing_plugin_data is not None:
                db_plugin_data = self._update_plugin_data_db(
                    existing_plugin_data=existing_plugin_data,
                    plugin_data=plugin_data,
                    session=session,
                )
            else:
                db_plugin_data = self._insert_plugin_data_db(
                    plugin_data=plugin_data,
                    session=session,
                )
        else:
            db_plugin_data = self._insert_plugin_data_db(
                plugin_data=plugin_data,
                session=session,
            )
        return schema.NendoPluginData.model_validate(db_plugin_data)

def add_related_collection(
    self,
    track_ids: List[Union[str, uuid.UUID]],
    collection_id: Union[str, uuid.UUID],
    name: str,
    description: str = "",
    user_id: Optional[Union[str, uuid.UUID]] = None,
    relationship_type: str = "relationship",
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoCollection:
    """Adds a new collection with a relationship to the given collection.

    Args:
        track_ids (List[Union[str, uuid.UUID]]): List of track ids.
        collection_id (Union[str, uuid.UUID]): Existing collection id.
        name (str): Name of the new related collection.
        description (str): Description of the new related collection.
        user_id (UUID, optional): The ID of the user adding the collection.
        relationship_type (str): Type of the relationship.
        meta (Dict[str, Any]): Meta of the new related collection.

    Returns:
        NendoCollection: The newly added NendoCollection object.
    """
    user_id = self._ensure_user_uuid(user_id)
    # Create a new collection
    new_collection = self.add_collection(
        name=name,
        user_id=user_id,
        track_ids=track_ids,
        description=description,
        collection_type=relationship_type,
        meta=meta,
    )

    with self.session_scope() as session:
        if isinstance(collection_id, str):
            collection_id = uuid.UUID(collection_id)
        collection = (
            session.query(model.NendoCollectionDB)
            .filter_by(id=collection_id)
            .first()
        )

        # Check if the collection does not exist
        if not collection:
            raise schema.NendoCollectionNotFoundError(
                "Collection not found",
                id=collection_id,
            )

        relationship = schema.NendoRelationshipCreate(
            source_id=new_collection.id,
            target_id=collection_id,
            relationship_type=relationship_type,
            meta=meta or {},
        )
        relationship = model.CollectionCollectionRelationshipDB(
            **relationship.model_dump(),
        )
        session.add(relationship)
    return new_collection

def add_related_track(
    self,
    file_path: Union[FilePath, str],
    related_track_id: Union[str, uuid.UUID],
    track_type: str = "str",
    user_id: Optional[Union[str, uuid.UUID]] = None,
    track_meta: Optional[Dict[str, Any]] = None,
    relationship_type: str = "relationship",
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoTrack:
    """Add a track from a file with a relationship to another track."""
    user_id = self._ensure_user_uuid(user_id=user_id)
    related_track_id = ensure_uuid(related_track_id)
    track = self.add_track(
        file_path=file_path,
        track_type=track_type,
        user_id=user_id,
        meta=track_meta,
    )
    relationship = schema.NendoRelationshipCreate(
        source_id=track.id,
        target_id=related_track_id,
        relationship_type=relationship_type,
        meta=meta or {},
    )
    with self.session_scope() as session:
        relationship = schema.NendoRelationship.model_validate(
            self._upsert_track_track_relationship(
                relationship=relationship,
                session=session,
            ),
        )
    return track

def add_related_track_from_signal(
    self,
    signal: np.ndarray,
    sr: int,
    related_track_id: Union[str, uuid.UUID],
    track_type: str = "track",
    user_id: Optional[Union[str, uuid.UUID]] = None,
    track_meta: Optional[Dict[str, Any]] = None,
    relationship_type: str = "relationship",
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoTrack:
    """Add a track from a signal with a relationship to another track."""
    user_id = self._ensure_user_uuid(user_id)
    related_track_id = ensure_uuid(related_track_id)
    track = self.add_track_from_signal(
        signal=signal,
        sr=sr,
        track_type=track_type,
        meta=track_meta,
    )
    relationship = schema.NendoRelationshipCreate(
        source_id=track.id,
        target_id=related_track_id,
        relationship_type=relationship_type,
        meta=meta or {},
    )
    with self.session_scope() as session:
        relationship = schema.NendoRelationship.model_validate(
            self._upsert_track_track_relationship(
                relationship=relationship,
                session=session,
            ),
        )
    return track

def add_track(
    self,
    file_path: Union[FilePath, str],
    track_type: str = "track",
    copy_to_library: Optional[bool] = None,
    skip_duplicate: Optional[bool] = None,
    user_id: Optional[uuid.UUID] = None,
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoTrack:
    """Add the track given by path to the library.

    Args:
        file_path (Union[FilePath, str]): Path to the file to be added.
        track_type (str, optional): Type of the track. Defaults to "track".
        copy_to_library (bool, optional): Flag that specifies whether
            the file should be copied into the library directory.
            Defaults to None.
        skip_duplicate (bool, optional): Flag that specifies whether a
            file should be added that already exists in the library, based on its
            file checksum. Defaults to None.
        user_id (UUID, optional): ID of user adding the track.
        meta (dict, optional): Metadata to attach to the track upon adding.

    Returns:
        schema.NendoTrack: The track that was added to the library.
    """
    skip_duplicate = skip_duplicate or self.config.skip_duplicate
    track = self._create_track_from_file(
        file_path=file_path,
        track_type=track_type,
        copy_to_library=copy_to_library,
        skip_duplicate=skip_duplicate,
        user_id=user_id or self.user.id,
        meta=meta,
    )
    if track is not None:
        with self.session_scope() as session:
            db_track = self._upsert_track_db(track=track, session=session)
            track = schema.NendoTrack.model_validate(db_track)
    return track

def add_track_from_signal(
    self,
    signal: np.ndarray,
    sr: int,
    track_type: str = "track",
    user_id: Optional[uuid.UUID] = None,
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoTrack:
    """Add a track to the library that is described by the given signal.

    Args:
        signal (np.ndarray): The numpy array containing the audio signal.
        sr (int): Sample rate
        track_type (str): Track type.
        user_id (UUID, optional): The ID of the user adding the track.
        meta (Dict[str, Any], optional): Track metadata. Defaults to {}.

    Returns:
        schema.NendoTrack: The added NendoTrack
    """
    track = self._create_track_from_signal(
        signal=signal,
        sr=sr,
        track_type=track_type,
        meta=meta,
        user_id=user_id or self.user.id,
    )
    if track is not None:
        with self.session_scope() as session:
            db_track = self._upsert_track_db(track=track, session=session)
            track = schema.NendoTrack.model_validate(db_track)
    return track

def add_track_relationship(
    self,
    track_one_id: Union[str, uuid.UUID],
    track_two_id: Union[str, uuid.UUID],
    relationship_type: str = "relationship",
    meta: Optional[Dict[str, Any]] = None,
) -> bool:
    """Add a relationship between two tracks."""
    track_one = self.get_track(track_one_id)  # track
    track_two = self.get_track(track_two_id)  # related

    # check that tracks are not the same
    if track_one_id == track_two_id:
        logger.error("Error must provide two different existing track ids")
        return False

    # check if tracks exist
    if track_one is None:
        logger.error("Error track id %s not found", str(track_one_id))
        return False

    if track_two is None:
        logger.error("Error track id %s not found", str(track_two_id))
        return False

    relationship = schema.NendoRelationshipCreate(
        source_id=track_one.id,
        target_id=track_two.id,
        relationship_type=relationship_type,
        meta=meta or {},
    )
    with self.session_scope() as session:
        relationship = schema.NendoRelationship.model_validate(
            self._upsert_track_track_relationship(
                relationship=relationship,
                session=session,
            ),
        )
    return True

def add_track_to_collection(
    self,
    track_id: Union[str, uuid.UUID],
    collection_id: Union[str, uuid.UUID],
    position: Optional[int] = None,
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoCollection:
    """Creates a relationship from the track to the collection.

    Args:
        track_id (Union[str, uuid.UUID]): ID of the track to add.
        collection_id (Union[str, uuid.UUID]): ID of the collection to
            which to add the track.
        position (int, optional): Target position of the track inside
            the collection.
        meta (Dict[str, Any]): Metadata of the relationship.

    Returns:
        NendoCollection: The updated NendoCollection object.
    """
    with self.session_scope() as session:
        # Convert IDs to UUIDs if they're strings
        collection_id = ensure_uuid(collection_id)
        track_id = ensure_uuid(track_id)

        # Check the collection and track objects
        collection = (
            session.query(model.NendoCollectionDB)
            .filter_by(id=collection_id)
            .first()
        )
        track = session.query(model.NendoTrackDB).filter_by(id=track_id).first()
        if not collection:
            raise schema.NendoCollectionNotFoundError(
                "The collection does not exist",
                collection_id,
            )
        if not track:
            raise schema.NendoCollectionNotFoundError(
                "The track does not exist",
                track_id,
            )

        rc_rel_db = model.TrackCollectionRelationshipDB
        if position is not None:
            # Update other states to keep ordering consistent
            session.query(rc_rel_db).filter(
                rc_rel_db.target_id == collection_id,
                rc_rel_db.relationship_position >= position,
            ).update(
                {
                    rc_rel_db.relationship_position: rc_rel_db.relationship_position
                    + 1,
                },
            )
        else:
            # If no position specified, add at the end
            last_state = (
                session.query(model.TrackCollectionRelationshipDB)
                .filter_by(target_id=collection_id)
                .order_by(
                    model.TrackCollectionRelationshipDB.relationship_position.desc(),
                )
                .first()
            )
            position = last_state.relationship_position + 1 if last_state else 0

        # Create a relationship from the track to the collection
        tc_relationship = model.TrackCollectionRelationshipDB(
            source_id=track_id,
            target_id=collection_id,
            relationship_type="track",
            meta=meta or {},
            relationship_position=position,
        )
        session.add(tc_relationship)
        session.commit()
        session.refresh(collection)
        return schema.NendoCollection.model_validate(collection)

def add_tracks(
    self,
    path: Union[str, DirectoryPath],
    track_type: str = "track",
    user_id: Optional[Union[str, uuid.UUID]] = None,
    copy_to_library: Optional[bool] = None,
    skip_duplicate: bool = True,
) -> schema.NendoCollection:
    """Scan the provided path and upsert the information into the library.

    Args:
        path (Union[str, DirectoryPath]): Path to the directory to be scanned
        track_type (str, optional): Track type for the new tracks
        user_id (UUID, optional): The ID of the user adding the track.
        copy_to_library (bool): Copy and convert the data into the nendo Library?
        skip_duplicate (bool): Skip adding duplicates?

    Returns:
        tracks (list[NendoTrack]): The tracks that were added to the Library
    """
    user_id = self._ensure_user_uuid(user_id)
    file_list = []
    if not os.path.exists(path):
        raise schema.NendoLibraryError(f"Source directory {path} does not exist.")
    for root, _, files in os.walk(path):
        file_list.extend(
            [
                os.path.join(root, file)
                for file in files
                if AudioFileUtils().is_supported_filetype(file)
            ],
        )
    tracks = self._add_tracks_db(
        file_paths=file_list,
        track_type=track_type,
        copy_to_library=copy_to_library,
        skip_duplicate=skip_duplicate,
        user_id=user_id,
    )
    return self.add_collection(name=path, track_ids=[t.id for t in tracks])

def add_tracks_to_collection(
    self,
    track_ids: List[Union[str, uuid.UUID]],
    collection_id: Union[str, uuid.UUID],
    meta: Optional[Dict[str, Any]] = None,
) -> schema.NendoCollection:
    """Creates a relationship from the track to the collection.

    Args:
        track_ids (List[Union[str, uuid.UUID]]): List of track ids to add.
        collection_id (Union[str, uuid.UUID]): ID of the collection to
            which to add the track.
        meta (Dict[str, Any], optional): Metadata of the relationship.

    Returns:
        NendoCollection: The updated NendoCollection object.
    """
    with self.session_scope() as session:
        # Convert IDs to UUIDs if they're strings
        collection_id = ensure_uuid(collection_id)
        track_ids = [ensure_uuid(track_id) for track_id in track_ids]

        # Check the collection and track objects
        collection = (
            session.query(model.NendoCollectionDB)
            .filter_by(id=collection_id)
            .first()
        )
        if not collection:
            raise schema.NendoCollectionNotFoundError(
                "The collection does not exist",
                collection_id,
            )
        existing_track_ids = (
            session.query(model.NendoTrackDB)
            .filter(model.NendoTrackDB.id.in_(track_ids))
            .all()
        )
        existing_track_ids = [t.id for t in existing_track_ids]
        missing_ids = [tid for tid in track_ids if tid not in existing_track_ids]
        if len(missing_ids) > 0:
            raise schema.NendoCollectionNotFoundError(
                "Tracks do not exist: ",
                missing_ids,
            )

        # append all tracks to the end of the collection
        last_postition = (
            session.query(model.TrackCollectionRelationshipDB)
            .filter_by(target_id=collection_id)
            .order_by(
                model.TrackCollectionRelationshipDB.relationship_position.desc(),
            )
            .first()
        )
        position = last_postition.relationship_position + 1 if last_postition else 0

        # create relationships from all tracks to the collection
        for track_id in track_ids:
            tc_relationship = model.TrackCollectionRelationshipDB(
                source_id=track_id,
                target_id=collection_id,
                relationship_type="track",
                meta=meta or {},
                relationship_position=position,
            )
            session.add(tc_relationship)
            position += 1
        session.commit()
        session.refresh(collection)
        return schema.NendoCollection.model_validate(collection)

@staticmethod
def batch_process(func):
    """Decorator to run functions multithreaded in batches.

    This decorator function transforms the given function to run
    in multiple threads. It expects that the first argument to the function
    is a list of items, which will be processed in parallel,
    in batches of a given size.
    """

    @functools.wraps(func)
    def wrapper(self, track=None, file_paths=None, *args, **kwargs):
        target = track or file_paths
        if isinstance(target, NendoTrack):
            return func(self, track=target, **kwargs)
        elif isinstance(target, list):  # noqa: RET505
            max_threads = self.config.max_threads
            batch_size = self.config.batch_size
            total = len(target)
            batches = [
                target[i : i + batch_size] for i in range(0, total, batch_size)
            ]
            start_time = time.time()
            futures = []

            def run_batch(batch_index, batch):
                try:
                    batch_start_time = time.time()
                    results = []
                    if track:
                        for _, item in enumerate(batch):
                            result = func(
                                self,
                                track=item,
                                *args,  # noqa: B026
                                **kwargs,
                            )
                            results.extend(result)
                    elif file_paths:
                        result = func(
                            self,
                            file_paths=batch,
                            *args,  # noqa: B026
                            **kwargs,
                        )
                        results.extend(result)
                    batch_end_time = time.time()
                    batch_time = time.strftime(
                        "%H:%M:%S",
                        time.gmtime(batch_end_time - batch_start_time),
                    )
                    total_elapsed_time = batch_end_time - start_time
                    average_time_per_batch = total_elapsed_time / (batch_index + 1)
                    estimated_total_time = average_time_per_batch * len(batches)
                    estimated_total_time_print = time.strftime(
                        "%H:%M:%S",
                        time.gmtime(estimated_total_time),
                    )
                    remaining_time = time.strftime(
                        "%H:%M:%S",
                        time.gmtime(estimated_total_time - total_elapsed_time),
                    )
                    logger.info(
                        f"Finished batch {batch_index + 1}/{len(batches)}.\n"
                        f"Time taken for this batch: {batch_time} - "
                        f"Estimated total time: {estimated_total_time_print} - "
                        f"Estimated remaining time: {remaining_time}\n",
                    )
                    return results
                except NendoError as e:
                    logger.exception(
                        "Error processing batch %d: %s",
                        batch_index,
                        e,
                    )

            with ThreadPoolExecutor(max_workers=max_threads) as executor:
                for batch_index, batch in enumerate(batches):
                    futures.append(executor.submit(run_batch, batch_index, batch))

            all_results = []
            for future in as_completed(futures):
                result = future.result()
                if result:
                    all_results.extend(future.result())
            return all_results
        else:
            raise TypeError("Expected NendoTrack or list of NendoTracks")

    return wrapper

def collection_size(
    self,
    collection_id: Union[str, uuid.UUID],
) -> int:
    """Get the number of tracks in a collection.

    Args:
        collection_id (Union[str, uuid.UUID]): The ID of the collection.

    Returns:
        int: The number of tracks.
    """
    collection_id = ensure_uuid(collection_id)
    with self.session_scope() as session:
        return (
            session.query(model.NendoTrackDB)
            .join(
                model.TrackCollectionRelationshipDB,
                model.TrackCollectionRelationshipDB.source_id
                == model.NendoTrackDB.id,
            )
            .filter(model.TrackCollectionRelationshipDB.target_id == collection_id)
            .count()
        )

def create_object(
    self,
    user_id: Optional[Union[str, uuid.UUID]] = None,
    track_type: str = "text",
    meta: Optional[Dict[str, Any]] = None,
    visibility: schema.Visibility = schema.Visibility.private,
    images: Optional[List[schema.NendoResource]] = None,
    file_path: str = "",
    resource_type: schema.ResourceType = schema.ResourceType.text,
    resource_meta: Optional[Dict[str, Any]] = None,
    copy_to_library: Optional[bool] = None,
) -> schema.NendoTrack:
    """Create a new object, manually.

    Args:
        user_id (uuid.UUID, optional): The ID of the object's user.
        track_type (str): The type of the object. Defaults to "track".
        meta (Dict[str, Any], optional): Metadata about the object.
        visibility (Visibility): Visibility of the object. Defaults to "private".
        images (List[NendoResource], optional): List of (additional) images for
            the object.
        file_path (str): Path to a file to be added to the track as NendoResource.
            Defaults to "" (no file path).
        resource_type (ResourceType): Type of the resource to set.
            Defaults to "text".
        resource_meta (dict, optional): Metadata about the NendoResource.
        copy_to_library (bool, optional): Flag that determines, whether to copy
            the file to the library storage or not.

    Returns:
        NendoTrack: The created track.
    """
    user_id = self._ensure_user_uuid(user_id)
    resource = None
    if len(file_path) > 0:
        copy_to_library = copy_to_library or self.config.copy_to_library
        if copy_to_library:
            try:
                # save file to storage in its original format
                path_in_library = self.storage_driver.save_file(
                    file_name=self.storage_driver.generate_filename(
                        filetype=os.path.splitext(file_path)[1][1:],  # without dot
                        user_id=str(user_id),
                    ),
                    file_path=file_path,
                    user_id=str(user_id),
                )
                location = self.storage_driver.get_driver_location()
            except Exception as e:  # noqa: BLE001
                raise schema.NendoLibraryError(
                    f"Error copying file to the library: {e}.",
                ) from e
        else:
            path_in_library = file_path
            location = schema.ResourceLocation.original

        resource = schema.NendoResource(
            file_path=self.storage_driver.get_file_path(
                src=path_in_library,
                user_id=str(user_id),
            ),
            file_name=self.storage_driver.get_file_name(
                src=path_in_library,
                user_id=str(user_id),
            ),
            resource_type=resource_type,
            location=location,
            meta=resource_meta or {},
        )
    else:
        location = self.storage_driver.get_driver_location()
        resource = schema.NendoResource(
            file_path="",
            file_name="",
            resource_type=resource_type,
            location=location,
            meta=resource_meta or {},
        )
    track = schema.NendoTrackCreate(
        user_id=user_id,
        visibility=visibility,
        images=images or [],
        resource=resource.model_dump(),
        track_type=track_type,
        meta=meta or {},
    )
    with self.session_scope() as session:
        db_track = self._upsert_track_db(track=track, session=session)
        return schema.NendoTrack.model_validate(db_track)

def export_collection(
    self,
    collection_id: Union[str, uuid.UUID],
    export_path: str,
    filename_suffix: str = "nendo",
    file_format: str = "wav",
) -> List[str]:
    """Export the collection to a directory.

    Args:
        collection_id (Union[str, uuid.UUID]): The ID of the target
            collection to export.
        export_path (str): Path to a directory into which the collection's tracks
            should be exported.
        filename_suffix (str): The suffix which should be appended to each
            exported track's filename.
        file_format (str, optional): Format of the exported track. Ignored if
            file_path is a full file path. Defaults to "wav".

    Returns:
        List[str]: A list with all full paths to the exported files.
    """
    collection_tracks = self.get_collection_tracks(collection_id)
    now = datetime.now().strftime("%Y%m%d%H%M%S")  # noqa: DTZ005
    if not os.path.isdir(export_path):
        logger.info(
            f"Export path {export_path} does not exist, creating now.",
        )
        os.makedirs(export_path, exist_ok=True)
    track_file_paths = []
    for track in collection_tracks:
        if track.has_meta("original_filename"):
            original_filename = track.get_meta("original_filename")
        else:
            original_filename = track.resource.file_name
        file_name = f"{original_filename}_{filename_suffix}_{now}.{file_format}"
        file_path = os.path.join(export_path, file_name)
        track_file_path = self.export_track(
            track_id=track.id,
            file_path=file_path,
            file_format=file_format,
        )
        track_file_paths.append(track_file_path)
    return track_file_paths

def export_track(
    self,
    track_id: Union[str, uuid.UUID],
    file_path: str,
    file_format: str = "wav",
) -> str:
    """Export the track to a file.

    Args:
        track_id (Union[str, uuid.UUID]): The ID of the target track to export.
        file_path (str): Path to the exported file. Can be either a full
            file path or a directory path. If a directory path is given,
            a filename will be automatically generated and the file will be
            exported to the format specified as file_format. If a full file
            path is given, the format will be deduced from the path and the
            file_format parameter will be ignored.
        file_format (str, optional): Format of the exported track. Ignored if
            file_path is a full file path. Defaults to "wav".

    Returns:
        str: The path to the exported file.
    """
    track = self.get_track(track_id=track_id)
    # Check if file_path is a directory
    if os.path.isdir(file_path):
        # Generate a filename with timestamp
        if track.has_meta("original_filename"):
            original_filename = track.get_meta("original_filename")
        else:
            original_filename = track.resource.file_name
        file_name = (
            f"{original_filename}_nendo_"
            f"{datetime.now().strftime('%Y%m%d%H%M%S')}"  # noqa: DTZ005
            f".{file_format}"
        )
        file_path = os.path.join(file_path, file_name)
    else:
        # Deduce file format from file extension
        file_format = os.path.splitext(file_path)[1].lstrip(".")

    # Exporting the audio
    temp_path = None
    signal = track.signal
    signal = np.transpose(signal) if signal.shape[0] <= 2 else signal
    if file_format in ("wav", "ogg"):
        sf.write(file_path, signal, track.sr, format=file_format)
    elif file_format == "mp3":
        # Create a temporary WAV file for conversion
        temp_path = file_path.rsplit(".", 1)[0] + ".wav"
        sf.write(temp_path, signal, track.sr)
        subprocess.run(
            ["ffmpeg", "-i", temp_path, "-acodec", "libmp3lame", file_path],
            check=False,
        )
    else:
        raise ValueError(
            "Unsupported file format. "
            "Supported formats are 'wav', 'mp3', and 'ogg'.",
        )

    # Clean up temporary file if used
    if temp_path and os.path.exists(temp_path):
        os.remove(temp_path)

    return file_path

def filter_related_tracks(
    self,
    track_id: Union[str, uuid.UUID],
    direction: str = "to",
    filters: Optional[Dict[str, Any]] = None,
    search_meta: Optional[Dict[str, Any]] = None,
    track_type: Optional[Union[str, List[str]]] = None,
    user_id: Optional[Union[str, uuid.UUID]] = None,
    collection_id: Optional[Union[str, uuid.UUID]] = None,
    plugin_names: Optional[List[str]] = None,
    order_by: Optional[str] = None,
    order: Optional[str] = "asc",
    limit: Optional[int] = None,
    offset: Optional[int] = None,
) -> Union[List, Iterator]:
    """Get tracks with a relationship to a track and filter the results.

    Args:
        track_id (Union[str, UUID]): ID of the track to be searched for.
        direction (str, optional): The relationship direction ("to", "from", "both").
        filters (dict, optional): Dictionary containing the filters to apply.
            Defaults to None.
        search_meta (dict, optional): Dictionary containing the keywords to search
            for over the track.resource.meta field. The dictionary's values
            should contain singular search tokens and the keys currently have no
            effect but might in the future. Defaults to {}.
        track_type (Union[str, List[str]], optional): Track type to filter for.
            Can be a singular type or a list of types. Defaults to None.
        user_id (Union[str, UUID], optional): The user ID to filter for.
        collection_id (Union[str, uuid.UUID], optional): Collection id to
            which the filtered tracks must have a relationship. Defaults to None.
        plugin_names (list, optional): List used for applying the filter only to
            data of certain plugins. If None, all plugin data related to the track
            is used for filtering.
        order_by (str, optional): Key used for ordering the results.
        order (str, optional): Order in which to retrieve results ("asc" or "desc").
        limit (int, optional): Limit the number of returned results.
        offset (int, optional): Offset into the paginated results (requires limit).

    Returns:
        Union[List, Iterator]: List or generator of tracks, depending on the
            configuration variable stream_mode
    """
    user_id = self._ensure_user_uuid(user_id)
    with self.session_scope() as session:
        query = self._get_related_tracks_query(
            track_id=ensure_uuid(track_id),
            session=session,
            direction=direction,
            user_id=user_id,
        )
        query = self._get_filtered_tracks_query(
            session=session,
            query=query,
            filters=filters,
            search_meta=search_meta,
            track_type=track_type,
            user_id=user_id,
            collection_id=collection_id,
            plugin_names=plugin_names,
        )
        return self.get_tracks(
            query=query,
            order_by=order_by,
            order=order,
            limit=limit,
            offset=offset,
            load_related_tracks=True,
            session=session,
        )