datasets

Dataset management: list, create, fetch, and delete datasets

cognee.datasets

Static class for managing datasets and their data. Methods that target a specific dataset identify it by UUID, never by name — see dataset name vs dataset id for how the two relate and which identifier each operation accepts.

For lower-level helpers such as get_dataset(), get_datasets_by_name(), and create_authorized_dataset(), see Dataset helper methods below.

Methods

datasets.list_datasets()

await cognee.datasets.list_datasets(user=None)

Returns all datasets accessible to the resolved user.

Parameter	Type	Default	Notes
user	Optional[User]	None	If omitted, Cognee resolves the default user.

datasets.discover_datasets()

cognee.datasets.discover_datasets(directory_path: str)

Discover dataset names from a local directory layout.

Parameter	Type	Default	Notes
directory_path	str	required	Local directory to scan for dataset-style subdirectories.

datasets.list_data()

await cognee.datasets.list_data(dataset_id, user=None)

Returns all Data records in a dataset.

This is the API to use when you want to read back DataItem fields stored during cognee.add(), such as label and external_metadata.

Parameter	Type	Default	Notes
dataset_id	UUID	required	Dataset UUID to inspect.
user	Optional[User]	None	If omitted, Cognee resolves the default user before permission checks.

datasets.has_data()

await cognee.datasets.has_data(dataset_id, user=None) -> bool

Check whether a dataset contains any data.

Parameter	Type	Default	Notes
dataset_id	str	required	Dataset identifier to check.
user	Optional[User]	None	If omitted, Cognee resolves the default user before permission checks.

datasets.get_status()

await cognee.datasets.get_status(
    dataset_ids: list[UUID],
    pipeline_names: list[str] | None = None,
) -> dict

Get pipeline status for one or more datasets.

When pipeline_names is omitted, this method keeps the legacy flat shape and returns the status of cognify_pipeline only.

Parameter	Type	Default	Notes
dataset_ids	list[UUID]	required	Dataset UUIDs to check.
pipeline_names	Optional[list[str]]	None	Pipeline names to query. If omitted, defaults to cognify_pipeline. Duplicate names are deduplicated while preserving order.

With no pipeline_names or a single pipeline name, the method returns {str(dataset_id): PipelineRunStatus}. With multiple pipeline names, it returns {str(dataset_id): {pipeline_name: PipelineRunStatus}}.

Possible values:

Value	Meaning
DATASET_PROCESSING_INITIATED	Pipeline queued but not yet started
DATASET_PROCESSING_STARTED	Pipeline is running
DATASET_PROCESSING_COMPLETED	Indexing finished successfully
DATASET_PROCESSING_ERRORED	Processing failed

Datasets with no recorded run for the requested pipeline are absent from the result.

status = await cognee.datasets.get_status([dataset.id])
# {"<dataset-uuid>": "DATASET_PROCESSING_COMPLETED"}

Troubleshooting UUID errors

get_status() expects dataset_ids to be a list of dataset UUIDs, not dataset names or string ids. Internally the values are bound against the pipeline_runs.dataset_id UUID column, so passing a plain string raises a SQLAlchemy StatementError wrapping one of:

• AttributeError: 'str' object has no attribute 'hex'
• ValueError: badly formed hexadecimal UUID string

# ❌ Wrong — passing a dataset name (or string id)
await cognee.datasets.get_status(["my_dataset"])

# ✅ Right — resolve the name to its UUID first
datasets = await cognee.datasets.list_datasets()
dataset_id = next(ds.id for ds in datasets if ds.name == "my_dataset")
status = await cognee.datasets.get_status([dataset_id])

If you already hold a string id (for example one read back from the HTTP API), wrap it in UUID before calling:

from uuid import UUID

status = await cognee.datasets.get_status([UUID(dataset_id_str)])

datasets.empty_dataset()

await cognee.datasets.empty_dataset(dataset_id, user=None)

Delete all data in a dataset and remove the dataset itself.

Parameter	Type	Default	Notes
dataset_id	UUID	required	Dataset UUID to empty.
user	Optional[User]	None	If omitted, Cognee resolves the default user and checks delete permission.

Notes

Despite the name, empty_dataset() does not leave an empty dataset record behind. It deletes graph content, data records, and the dataset entity itself.

datasets.delete_data()

await cognee.datasets.delete_data(
    dataset_id,
    data_id,
    user=None,
    mode="soft",
    delete_dataset_if_empty=False,
)

Delete a specific data item from a dataset.

Parameter	Type	Default	Notes
dataset_id	UUID	required	Dataset UUID containing the target data item.
data_id	UUID	required	Data item UUID to delete.
user	Optional[User]	None	If omitted, Cognee resolves the default user and checks delete permission.
mode	str	soft	Kept for backward compatibility. The implementation warns against using "hard".
delete_dataset_if_empty	bool	False	If True, deletes the dataset when the removed item was its last remaining data item.

Notes

mode="hard" is preserved for backward compatibility, but the implementation explicitly warns not to use it.

What delete_data() removes across stores

delete_data() is not a relational-only operation. It cleans up every backend that holds memory derived from the targeted data item:

Store	What is removed
Graph (Kuzu, Neo4j, Neptune, FalkorDB, …)	The nodes and edges Cognee recorded as derived from that data item.
Vector (LanceDB, Qdrant, PGVector, …)	The matching points in every {NodeType}_{indexed_field} collection (for example Entity_name, DocumentChunk_text), plus the EdgeType_relationship_name and Triplet_text entries belonging to the removed edges.
Relational (SQLite, Postgres)	The per-document node/edge ownership records and the Data row itself. A Data row belongs to exactly one dataset, so there is no membership to unlink and nothing to keep alive on behalf of another dataset — the identical content in another dataset is a separate row that is left untouched.
File storage (local disk, S3)	The raw file — but only when this was the last Data row (in any dataset) pointing at that location and the file lives under DATA_ROOT_DIRECTORY (that is, Cognee copied it in). Files referenced from outside that directory are left in place.
Session cache (SQL, Redis, filesystem)	Only the contaminated entries in sessions attributed to this dataset: turns whose recorded graph elements overlap the nodes and edges just deleted, plus what they propagated to — feedback referencing such a turn, the distilled session-context lesson it fed, and later turns that consumed that lesson. Untouched turns, and the sessions themselves, are kept.

Session cleanup is best-effort: a cache failure is logged as a warning and never fails delete_data(), so a {"status": "success"} result does not guarantee every entry was removed. It is also the fine-grained variant — empty_dataset() deletes every session attributed to the dataset outright, and forget(everything=True) prunes the cache wholesale.

What is intentionally left behind:

• Shared nodes. An entity such as "New York" that another data item also references stays in the graph and vector store; only nodes unique to the deleted item are dropped. Relationships between surviving shared nodes are not deleted either.
• The dataset. The dataset record survives unless you pass delete_dataset_if_empty=True and the removed item was the last one.

See Delete for the same flow in narrative form, and forget() for a scope matrix across all deletion modes.

Cascading changes when a source document changes

delete_data() only removes; it never re-extracts. To propagate an edit to a source document through the graph and vector stores, use update(), which calls delete_data() for the old item, re-adds the new content, and re-runs cognify on the dataset:

await cognee.update(
    data_id=item.id,
    data="Updated document content.",
    dataset_id=ds.id,
)

Relationships that existed only in the old version disappear with the deleted nodes; relationships found in the new content are created by the cognify step. incremental_loading=True (the default) keeps the other, unchanged documents in the dataset from being reprocessed — pass incremental_loading=False only when the whole dataset should be rebuilt, for example after changing your graph model or prompts.

To clear a document’s derived memory while keeping its record and raw file, use forget(..., memory_only=True) and re-run cognify.

datasets.delete_all()

await cognee.datasets.delete_all(user=None)

Delete all datasets the user has permission to delete.

Parameter	Type	Default	Notes
user	Optional[User]	None	If omitted, Cognee resolves the default user.

Dataset helper methods

cognee.datasets covers the common cases. Underneath it, Cognee exports a set of dataset helpers importable from cognee.modules.data.methods. Use them when you need to resolve a dataset by name or id, create one explicitly, or apply a permission type other than read. All of them are async except check_dataset_name().

Fetching datasets

Ownership-scoped lookups (they match on Dataset.owner_id only, ignoring ACLs):

Method	Signature	Returns
get_dataset()	get_dataset(user_id: UUID, dataset_id: UUID)	The Dataset, or None if it does not exist or is not owned by user_id.
get_datasets()	get_datasets(user_id: UUID)	All Dataset rows owned by user_id.
get_datasets_by_name()	get_datasets_by_name(dataset_names: str | list[str], user_id: UUID)	Datasets owned by user_id whose name is in dataset_names. A single string is treated as a one-item list.
get_dataset_data()	get_dataset_data(dataset_id: UUID)	The Data records in a dataset, largest first. Backs datasets.list_data().
has_dataset_data()	has_dataset_data(dataset_id: UUID)	True if the dataset has at least one data record. Backs datasets.has_data().

Permission-aware lookups (they take a User object and go through the permissions system, so they also return datasets shared with the user):

Method	Signature	Returns
get_authorized_dataset()	get_authorized_dataset(user: User, dataset_id: UUID, permission_type: str = "read")	The Dataset if the user holds permission_type on it, otherwise None.
get_authorized_dataset_by_name()	get_authorized_dataset_by_name(dataset_name: str, user: User, permission_type: str)	The first authorized dataset with that name, otherwise None.
get_authorized_existing_datasets()	get_authorized_existing_datasets(datasets: list[str] | list[UUID] | None, permission_type: str, user: User)	All datasets the user holds permission_type on, filtered to datasets when that argument is non-empty. Backs datasets.list_datasets().
get_dataset_ids()	get_dataset_ids(datasets: list[str] | list[UUID], user: User)	Dataset UUIDs for the given identifiers. Names are only resolved against datasets the user owns in their tenant — to target a dataset owned by someone else, pass its UUID. Raises DatasetTypeError on mixed or unsupported types.

permission_type is one of read, write, delete, or share.

Creating and deleting datasets

Method	Signature	Notes
create_dataset()	create_dataset(dataset_name: str, user: User)	Returns the existing dataset with that name for the user’s owner/tenant pair, or creates it. The id is derived deterministically from the name and owner, so different users can reuse the same dataset name. Grants no permissions.
create_authorized_dataset()	create_authorized_dataset(dataset_name: str, user: User)	create_dataset() plus read, write, delete, and share permissions for user — and for their parent user when parent_user_id is set. Use this one in multi-user setups.
load_or_create_datasets()	load_or_create_datasets(dataset_names: list[str | UUID], existing_datasets: list[Dataset], user: User)	Reuses matching datasets from existing_datasets and creates the rest via create_authorized_dataset(). Raises DatasetNotFoundError if a UUID has no match.
delete_dataset()	delete_dataset(dataset: Dataset)	Deletes the dataset row, its dedicated graph and vector databases if it has any, and the Data rows the dataset owns. Takes a Dataset object, not an id.
check_dataset_name()	check_dataset_name(dataset_name: str)	Synchronous. Raises ValueError if the name contains a space or a dot.

Because a Data row belongs to exactly one dataset, delete_dataset() also deletes that dataset’s Data rows. Each removal goes through the same raw-file rule as delete_data(): the file on disk is only removed when no other Data row still points at that raw_data_location and it lives under DATA_ROOT_DIRECTORY. Use datasets.empty_dataset() when you want to clear a dataset’s data but keep the dataset itself.

Examples

Basic dataset operations

import cognee

# List all datasets
datasets = await cognee.datasets.list_datasets()
for ds in datasets:
    print(ds.name, ds.id)

# Check dataset contents
data = await cognee.datasets.list_data(dataset_id=ds.id)

# Delete a specific item
await cognee.datasets.delete_data(
    dataset_id=ds.id,
    data_id=item.id,
)

# Wipe everything
await cognee.datasets.delete_all()

Poll for indexing completion across parallel datasets

Use get_status() in a wait loop to confirm all datasets in a parallel batch have finished indexing before querying.

import asyncio
import cognee
from cognee.modules.pipelines.models import PipelineRunStatus

TERMINAL = {
    PipelineRunStatus.DATASET_PROCESSING_COMPLETED,
    PipelineRunStatus.DATASET_PROCESSING_ERRORED,
}

async def wait_for_indexing(dataset_ids, poll_interval=3, timeout=120):
    for _ in range(timeout // poll_interval):
        statuses = await cognee.datasets.get_status(dataset_ids)
        if all(s in TERMINAL for s in statuses.values()):
            return statuses
        await asyncio.sleep(poll_interval)
    raise TimeoutError("Indexing did not complete in time")

async def main():
    batches = {
        "batch_a": ["doc1.pdf", "doc2.pdf"],
        "batch_b": ["doc3.pdf", "doc4.pdf"],
    }

    # Add and index multiple datasets in parallel
    await asyncio.gather(*[
        cognee.add(files, dataset_name=name) for name, files in batches.items()
    ])
    await asyncio.gather(*[
        cognee.cognify(datasets=[name]) for name in batches
    ])

    # Confirm all datasets reached a terminal status
    all_datasets = await cognee.datasets.list_datasets()
    dataset_ids = [ds.id for ds in all_datasets if ds.name in batches]
    statuses = await wait_for_indexing(dataset_ids)

    for ds_id, status in statuses.items():
        if status == PipelineRunStatus.DATASET_PROCESSING_COMPLETED:
            print(f"{ds_id}: indexed successfully")
        else:
            print(f"{ds_id}: error — {status.value}")

asyncio.run(main())

The same pattern works when indexing is triggered via the HTTP API — poll get_status() from a separate process until all datasets reach DATASET_PROCESSING_COMPLETED or DATASET_PROCESSING_ERRORED.

Read back DataItem metadata

import cognee
from cognee.tasks.ingestion.data_item import DataItem

await cognee.add(
    DataItem(
        "/path/to/report.pdf",
        label="q4-report",
        external_metadata={"author": "Jane Smith", "quarter": "Q4-2024"},
    ),
    dataset_name="finance",
)

datasets = await cognee.datasets.list_datasets()
data_items = await cognee.datasets.list_data(dataset_id=datasets[0].id)

for item in data_items:
    print(item.label, item.external_metadata)
    # q4-report  {"author": "Jane Smith", "quarter": "Q4-2024"}

external_metadata is stored on the relational Data record only. It is not placed into the vector store or knowledge graph and is not returned by cognee.search(). If you need metadata to be vector-searchable, define a custom DataPoint subclass and list the fields to embed in metadata.index_fields. See DataPoints.