cognee.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, newest first (created_at descending, ties broken by id).

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

The method itself takes no paging arguments and always returns the complete set, whatever its size. How it gets there differs by mode: locally it reads every row in one query, with no pagination or offset limit in play, while in remote mode it walks the server’s bounded pages for you — see below.

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. Not used in remote mode — see below.

Remote mode (serve())

While a serve() connection is active, list_data() reads the remote dataset over GET /api/v1/datasets/{dataset_id}/data instead of the local store, so it works for datasets that exist only on the remote instance. The user argument is ignored on that path: the API key used by serve() identifies the caller, and the server runs its own permission check.

That endpoint is paginated, so the remote path requests pages of 1000 rows and advances until a short page comes back, deduplicating rows on id in case a concurrent insert or delete shifts them between pages. The return value is still the whole dataset, but it is assembled from several round trips rather than one — reading a large dataset over a connection costs one request per 1000 rows. A server that does not honor the request raises RuntimeError rather than looping: a response that is not a list, and one that comes back longer than the requested page or contains only rows already seen. Any 4xx/5xx raises too — including the 422 the server returns once traversal passes offset 1,000,000, which puts a dataset of 1,001,000 rows or more out of reach of this method (at exactly that size the last page comes back full, so the traversal asks for one more and takes the 422 instead of returning). If you only need a total, call GET /api/v1/datasets/{dataset_id}/data/count instead of listing the rows.

Remote rows are parsed back through DataDTO, the same wire model the server serializes them with, so the fields it returns carry the same attribute names and types as their local counterparts — row.id is a UUID (not a string), and the field is row.mime_type (not camelCase mimeType). label and external_metadata survive the round trip, so the DataItem fields stored during cognee.add() are readable remotely too.

A remote row exposes exactly these fields: id, name, created_at, updated_at, extension, mime_type, raw_data_location, dataset_id, label, external_metadata, and data_size (the stored file size in bytes, None for a row written without one). created_at is also optional and can be None for rows written by a writer that skipped the ORM default. A local Data record carries more — content_hash, owner_id, tenant_id, loader_engine, system_metadata, and others — so code that reads those attributes works locally and raises AttributeError over a connection.

list_datasets() is not routed to the remote instance — it still reads the local store. The common pattern of listing datasets and then reading one’s data breaks over a serve() connection, because the dataset ids come from the local store and the data lookup goes to the remote one. Get remote dataset ids from the instance itself (GET /api/v1/datasets) or from the Cognee Cloud UI, and pass them to list_data() directly.

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	Legacy value, no longer written; only rows created by older Cognee versions carry it
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"}

get_status() reports the state of a dataset’s pipeline run. For the state of its individual items — which data items carry the per-item completion stamp for a pipeline — use the HTTP endpoint GET /api/v1/datasets/{dataset_id}/processing-status. There is no SDK method for it.

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, across both the sessions attributed to this dataset and the sessions carrying no dataset attribution (the global default_session an unscoped or cross-dataset recall runs in): 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 only then applies this same targeted pass to unattributed sessions), 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 by default diffs the new content against the stored text and replaces only the chunks the edit touched — falling back to a full rebuild that drops the document’s memory with forget(memory_only=True), refreshes the same record with the new content, and re-runs cognify on the dataset when the chunk-level preconditions are not met:

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, limit: int | None = None, offset: int = 0, *, order_by: "size" | "created_at" = "size")	The Data records in a dataset. Unbounded by default — limit=None returns every row, which is what internal pipeline callers need; the HTTP route passes a bounded limit of its own. order_by defaults to "size" (largest first, for pipeline batching); "created_at" is newest-first with id as tiebreak, the order datasets.list_data() and GET /api/v1/datasets/{dataset_id}/data use. Any other value raises ValueError. Backs datasets.list_data().
count_dataset_data()	count_dataset_data(dataset_id: UUID)	Number of Data records in a dataset, counted in the database instead of by fetching the rows. Backs GET /api/v1/datasets/{dataset_id}/data/count.
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, strict: bool = False)	All datasets the user holds permission_type on, filtered to datasets when that argument is non-empty. strict is passed straight through to get_dataset_ids(). Backs datasets.list_datasets().
get_dataset_ids()	get_dataset_ids(datasets: list[str] | list[UUID], user: User, strict: bool = False)	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. With strict=True, a name that resolves to nothing raises DatasetNotFoundError (404) listing every miss (duplicates once) instead of being dropped; search() and recall() use this, write paths keep the lenient default so an unmatched name can still mean “create it”. 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.