Capsule 05: Pinecone namespaces — native multi-tenant isolation, and why it beats metadata filtering

Capsule description

In M06 you implemented multi-tenant isolation with where={"workspace_id": "X"}. It works — as long as the filter is applied correctly. The problem: in large systems, one new endpoint that forgets the filter is all it takes to have a data leak. The defense is code + tests + linters, all correct, but all of it defense by convention.

Pinecone has a structural alternative: namespaces. Each tenant lives in a separate logical space inside the same index. A query to namespace="acme" physically CANNOT see vectors from namespace="other_tenant" — the separation is in the engine, not in the filter. It's the equivalent of having separate databases per tenant while paying for only one.

This capsule teaches you how to use namespaces as your primary isolation layer, how to combine them with metadata filtering for sub-segmentation within the tenant, and why for production multi-tenant systems it's always the right choice.

By the end of this capsule you'll be able to:

• ✅ Explain the conceptual difference between a namespace and a metadata filter
• ✅ Implement a wrapper that forces the right namespace per tenant
• ✅ Combine namespaces (tenant) + metadata filters (sub-segmentation)
• ✅ Design isolation tests that physically validate the separation
• ✅ Decide when to use a namespace, metadata, or both
• ✅ Anticipate the operational traps: a nonexistent namespace, inconsistent naming, badly designed cross-namespace queries

Estimated time: 30-35 minutes

The conceptual difference: filter vs namespace

                        APPROACH 1: METADATA FILTER (M06)

     Pinecone Index
     ┌──────────────────────────────────────────┐
     │  Vectors from ALL tenants, mixed together│
     │                                           │
     │  vec1 (tenant=A)  vec2 (tenant=B)  vec3  │ ← all in the same space
     │  vec4 (tenant=A)  vec5 (tenant=C)  ...   │
     │                                           │
     │  Query: filter={"tenant": "A"}            │
     │     ↓                                     │
     │  Pinecone searches ALL the vectors,       │
     │  then discards the ones that aren't A     │
     │                                           │
     │  If the filter fails → DATA LEAK          │
     └──────────────────────────────────────────┘


                        APPROACH 2: NAMESPACES

     Pinecone Index
     ┌────────────────┬────────────────┬─────────────────┐
     │  namespace=A   │  namespace=B   │  namespace=C    │
     │                │                │                  │
     │  vec1, vec4    │  vec2          │  vec5           │
     │  ...           │  ...           │  ...            │
     └────────────────┴────────────────┴─────────────────┘

     A query with namespace="A":
       → Pinecone searches ONLY namespace=A
       → Physically impossible to see namespace=B/C
       → A data leak is impossible even if the code has a bug

     Same index, same cost, separation guaranteed by the engine

The key difference: with a filter, security depends on the code always including the right filter. With a namespace, security depends on the engine. If the query runs with no namespace or with the wrong namespace, it returns empty — it can't leak data.

The correct implementation

Pattern 1: a single function to build the namespace

# pinecone_namespaces.py


def tenant_namespace(workspace_id: str) -> str:
    """
    The single convention for a workspace's namespace.
    Use it EVERYWHERE in the code to avoid inconsistencies.
    """
    if not workspace_id:
        raise ValueError("workspace_id is required")

    # Sanitize: lowercase, no odd characters
    sanitized = workspace_id.lower().replace(" ", "-").strip()

    return f"ws-{sanitized}"

Why a single function:

• If you later decide to change the convention (e.g. adding an environment prefix), there's one place to do it.
• It's impossible for one dev to write f"workspace-{wid}" and another f"ws-{wid}" — everyone uses the same function.

Pattern 2: secure_query with a mandatory namespace

# secure_query_pinecone.py
from pinecone import Pinecone, Index


class TenantIsolationError(Exception):
    pass


def secure_query_pinecone(
    index: Index,
    query_vector: list[float],
    tenant: TenantContext,
    additional_filters: dict = None,
    top_k: int = 5,
):
    """
    A wrapper that ALWAYS uses the right namespace.
    Impossible to skip this wrapper without getting namespace=None → an invalid query.
    """
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("workspace_id is mandatory")

    namespace = tenant_namespace(tenant.workspace_id)

    response = index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,                    # ← mandatory
        filter=additional_filters,              # optional, for sub-segmentation
        include_metadata=True,
    )

    # Audit log
    log_query_audit(tenant, namespace, top_k)

    return response


def secure_upsert_pinecone(
    index: Index,
    vectors: list,
    tenant: TenantContext,
):
    """A wrapper for upserts with the namespace forced in."""
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("workspace_id is mandatory")

    namespace = tenant_namespace(tenant.workspace_id)
    return index.upsert(vectors=vectors, namespace=namespace)

Pattern 3: automated isolation tests

# tests/test_namespace_isolation.py
import pytest
from secure_query_pinecone import secure_query_pinecone, TenantIsolationError


@pytest.fixture
def index_with_two_tenants():
    """Setup with vectors in different namespaces."""
    index = pinecone_client.Index("test-isolation")

    # Tenant A: 100 vectors
    vectors_a = [
        {"id": f"a_{i}", "values": [0.1] * 1536, "metadata": {"text": f"doc A {i}"}}
        for i in range(100)
    ]
    index.upsert(vectors=vectors_a, namespace="ws-tenant-a")

    # Tenant B: 100 vectors
    vectors_b = [
        {"id": f"b_{i}", "values": [0.2] * 1536, "metadata": {"text": f"doc B {i}"}}
        for i in range(100)
    ]
    index.upsert(vectors=vectors_b, namespace="ws-tenant-b")

    return index


def test_tenant_a_cannot_see_namespace_b(index_with_two_tenants):
    """Tenant A can NOT see tenant B's vectors."""
    ctx = TenantContext(workspace_id="tenant-a", user_id="u1")
    response = secure_query_pinecone(
        index_with_two_tenants,
        query_vector=[0.1] * 1536,
        tenant=ctx,
        top_k=20,
    )

    for match in response["matches"]:
        assert match["id"].startswith("a_"), (
            f"LEAK: tenant-a saw {match['id']} (expected only a_*)"
        )


def test_nonexistent_namespace_returns_empty(index_with_two_tenants):
    """A query to a namespace that doesn't exist does NOT leak — it returns empty."""
    ctx = TenantContext(workspace_id="tenant-c", user_id="u1")  # tenant-c has no vectors
    response = secure_query_pinecone(
        index_with_two_tenants,
        query_vector=[0.1] * 1536,
        tenant=ctx,
        top_k=20,
    )

    assert len(response["matches"]) == 0, (
        f"LEAK: a nonexistent namespace returned matches: {response['matches']}"
    )


def test_workspace_id_is_required(index_with_two_tenants):
    """Without a workspace_id you can't run a query."""
    with pytest.raises(TenantIsolationError):
        secure_query_pinecone(
            index_with_two_tenants,
            query_vector=[0.1] * 1536,
            tenant=None,
            top_k=5,
        )


def test_concurrent_tenants_no_leakage(index_with_two_tenants):
    """Concurrent queries from different tenants are isolated."""
    import threading

    leaks = []

    def query_as_tenant(tenant_id: str):
        ctx = TenantContext(workspace_id=tenant_id, user_id=f"u_{tenant_id}")
        response = secure_query_pinecone(
            index_with_two_tenants,
            query_vector=[0.1] * 1536,
            tenant=ctx,
            top_k=20,
        )
        for match in response["matches"]:
            expected_prefix = "a_" if tenant_id == "tenant-a" else "b_"
            if not match["id"].startswith(expected_prefix):
                leaks.append((tenant_id, match["id"]))

    threads = [
        threading.Thread(target=query_as_tenant, args=("tenant-a",))
        for _ in range(10)
    ] + [
        threading.Thread(target=query_as_tenant, args=("tenant-b",))
        for _ in range(10)
    ]

    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not leaks, f"Concurrent leakage: {leaks}"

Combining namespaces + metadata filtering

A namespace for the tenant. Metadata for everything else (visibility, type, date, tags).

def search_with_full_security(
    index: Index,
    query_vector: list[float],
    tenant: TenantContext,
    optional_filters: dict = None,
    top_k: int = 5,
):
    """
    The complete security pipeline:
    - Namespace: tenant isolation (the engine)
    - Filter: visibility, type, date, tags (the metadata)
    """
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("workspace_id is mandatory")

    namespace = tenant_namespace(tenant.workspace_id)

    # Build the visibility filter based on role
    visibility_filter = build_visibility_filter(tenant, optional_filters)

    return index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,           # ← the primary isolation
        filter=visibility_filter,      # ← the sub-segmentation
        include_metadata=True,
    )


def build_visibility_filter(tenant: TenantContext, additional: dict = None):
    """Builds the metadata filter for visibility, based on role."""
    visibility_clauses = []

    # Public to the tenant
    visibility_clauses.append({"visibility": "public"})

    # Team docs if the user is on the team
    if tenant.project_ids:
        visibility_clauses.append({
            "$and": [
                {"visibility": "team"},
                {"project_id": {"$in": tenant.project_ids}},
            ]
        })

    # Personal, the owner's only
    visibility_clauses.append({
        "$and": [
            {"visibility": "personal"},
            {"owner_id": {"$eq": tenant.user_id}},
        ]
    })

    # Admin docs only if they're an admin
    if tenant.user_role in ("admin", "owner"):
        visibility_clauses.append({"visibility": "admin_only"})

    base_filter = {"$or": visibility_clauses}

    if additional:
        return {"$and": [base_filter, additional]}
    return base_filter

The result: two layers of security. Even if the visibility filter has a bug, the namespace guarantees a minimum of isolation between tenants.

When to use a namespace, metadata, or both

The rule: a namespace for the strongest isolation dimension (typically the tenant). Metadata for sub-segmentation within it.

Advanced patterns

Pattern A: two levels of namespace

For large corpora with many tenants and projects:

def project_namespace(workspace_id: str, project_id: str) -> str:
    """A fine-grained namespace: tenant + project."""
    return f"ws-{workspace_id}__proj-{project_id}"


# Usage
namespace = project_namespace("acme", "marketing")
# Result: "ws-acme__proj-marketing"

The trade-off: better isolation, but cross-project queries get more complex (you need multiple namespaces).

Pattern B: a namespace per environment

def env_namespace(workspace_id: str, env: str = None) -> str:
    """A namespace that includes the environment, for staging/prod separation."""
    env = env or os.getenv("APP_ENV", "prod")
    return f"{env}__ws-{workspace_id}"

Useful when a tenant has staging data that must NOT be mixed with prod.

Pattern C: listing the existing namespaces

def list_active_tenants(index: Index) -> list[str]:
    """Lists every tenant with data in the index."""
    stats = index.describe_index_stats()
    namespaces = stats["namespaces"]

    return [
        ns_name.replace("ws-", "")
        for ns_name in namespaces.keys()
        if ns_name.startswith("ws-") and namespaces[ns_name]["vector_count"] > 0
    ]


# Useful for admin operations (analytics, audit, cleanup)
active_tenants = list_active_tenants(index)
print(f"Active tenants: {len(active_tenants)}")

Traps and common mistakes

Trap 1: forgetting the namespace in the query

The mistake:

response = index.query(vector=v, top_k=5)  # ← no namespace

The symptom: Pinecone searches the "" (default) namespace. If nobody wrote there, it returns empty. But if someone stored data in the default by mistake, that's a leak between tenants.

How to prevent it: ALWAYS use secure_query_pinecone, which enforces the namespace.

Trap 2: inconsistent naming across devs

The mistake: dev A writes f"workspace-{wid}", dev B writes f"ws-{wid}". The same tenant has data in two different namespaces.

The symptom: dev A's queries don't see the docs dev B ingested. "Fragmented" data.

How to prevent it: a single tenant_namespace() function used EVERYWHERE in the code.

Trap 3: a test's namespace left behind in production

The mistake: a dev runs a test in production with namespace="my-test". The test ends, and the namespace stays around full of junk.

The symptom: after months, you have 50 namespaces from old tests. Auditors see docs with no clear owner.

How to prevent it:

• Tests always in the staging index, never prod.
• If tests in prod are unavoidable, use the namespace prefix test- and a periodic cleanup script.

Trap 4: badly designed cross-namespace queries

The mistake: an admin needs statistics across every tenant. They loop over all the namespaces:

for ns in active_tenants:
    response = index.query(vector=v, top_k=5, namespace=ns)
    # ... process ...

The symptom: it works but it's slow (N queries) and can hit the rate limits.

How to prevent it: for admin operations that need cross-tenant data, consider:

• Aggregated materialized views (computed offline).
• A separate index for approved cross-tenant data.
• Selective queries (only the top tenants instead of all of them).

Trap 5: a namespace with invalid characters

The mistake: tenant_namespace("Acme Corp / Brazil") → "ws-Acme Corp / Brazil".

The symptom: Pinecone may accept it (with limitations) or reject it depending on the version.

How to prevent it: the namespace function sanitizes:

import re

def tenant_namespace(workspace_id: str) -> str:
    sanitized = re.sub(r'[^a-z0-9-_]', '-', workspace_id.lower())
    return f"ws-{sanitized}"

Trap 6: thinking namespaces are free

The mistake: creating a namespace per user (instead of per tenant).

The symptom: an index with 1M namespaces. The performance of describe_index_stats degrades.

How to prevent it: namespaces are tenant-level granularity (typically 50-10K). For finer granularity, a metadata filter is better.

Applied exercise

Scenario: you're an AI Engineer at a marketing SaaS company. The data:

• 200 customers (tenants), each with 1-50 projects
• Total: ~5000 projects
• Each project has 5K-100K docs
• Compliance: GDPR + a SOC2 audit every 6 months
• Typical queries: within a specific project

Your job:

1. Decide the namespace strategy (one per customer, or one per project).
2. Design an appropriate secure_query and secure_upsert.
3. A plan for automated isolation tests in CI.

# secure_marketing_search.py
from typing import Literal


def tenant_namespace(workspace_id: str) -> str:
    sanitized = workspace_id.lower().replace(" ", "-").strip()
    return f"ws-{sanitized}"


@dataclass(frozen=True)
class MarketingTenantContext:
    workspace_id: str
    user_id: str
    user_role: Literal["admin", "manager", "team_member"]
    project_ids: list[str]  # the projects the user belongs to

    def __post_init__(self):
        if not self.workspace_id or not self.user_id:
            raise TenantIsolationError("workspace_id and user_id are mandatory")


def secure_marketing_query(
    index,
    query_vector: list[float],
    tenant: MarketingTenantContext,
    project_id: str = None,
    additional_filters: dict = None,
    top_k: int = 5,
):
    """
    The security pipeline:
    - Namespace: isolation per tenant
    - Filter: project_id (if passed) + visibility based on role
    """
    namespace = tenant_namespace(tenant.workspace_id)

    # Validate access to the project
    if project_id and project_id not in tenant.project_ids:
        raise PermissionError(f"The user does not belong to project {project_id}")

    # Build the filter
    filter_clauses = []

    if project_id:
        # A query into a specific project
        filter_clauses.append({"project_id": {"$eq": project_id}})
    else:
        # A query across all the user's projects (the manager view)
        filter_clauses.append({"project_id": {"$in": tenant.project_ids}})

    # Visibility based on role
    if tenant.user_role == "team_member":
        filter_clauses.append({"visibility": {"$ne": "admin_only"}})
    # admin and manager see everything

    if additional_filters:
        filter_clauses.append(additional_filters)

    final_filter = {"$and": filter_clauses}

    return index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,
        filter=final_filter,
        include_metadata=True,
    )


def secure_marketing_upsert(
    index,
    vectors: list,
    tenant: MarketingTenantContext,
):
    """An upsert with the namespace forced in."""
    namespace = tenant_namespace(tenant.workspace_id)

    # Validate that every vector has a project_id in its metadata
    for v in vectors:
        if "project_id" not in v["metadata"]:
            raise ValueError("Every vector must have a project_id in its metadata")
        if v["metadata"]["project_id"] not in tenant.project_ids:
            raise PermissionError(
                f"The user does not belong to project {v['metadata']['project_id']}"
            )

    return index.upsert(vectors=vectors, namespace=namespace)

# tests/test_marketing_isolation.py
import pytest


@pytest.fixture
def index_multi_tenant():
    """Setup: 3 tenants × 2 projects each."""
    index = pinecone_client.Index("test-marketing")

    for tenant in ["acme", "globex", "wonka"]:
        for project in ["mkt", "sales"]:
            vectors = [
                {
                    "id": f"{tenant}_{project}_{i}",
                    "values": [0.1] * 1536,
                    "metadata": {
                        "project_id": f"{tenant}_{project}",
                        "visibility": "public",
                    },
                }
                for i in range(100)
            ]
            index.upsert(vectors=vectors, namespace=f"ws-{tenant}")

    return index


def test_cross_tenant_isolation(index_multi_tenant):
    """Acme can NOT see Globex's docs."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="manager",
        project_ids=["acme_mkt", "acme_sales"],
    )

    response = secure_marketing_query(
        index_multi_tenant,
        [0.1] * 1536,
        ctx,
        top_k=50,
    )

    for match in response["matches"]:
        assert match["id"].startswith("acme_"), (
            f"LEAK: acme saw {match['id']}"
        )


def test_cross_project_within_tenant(index_multi_tenant):
    """An Acme team_member only on the mkt project does NOT see sales."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="team_member",
        project_ids=["acme_mkt"],  # mkt ONLY
    )

    response = secure_marketing_query(
        index_multi_tenant,
        [0.1] * 1536,
        ctx,
        top_k=50,
    )

    for match in response["matches"]:
        assert "_mkt_" in match["id"], (
            f"LEAK: a user only on mkt saw {match['id']}"
        )


def test_cannot_query_unauthorized_project():
    """A user can't pass the project_id of a project they don't belong to."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="team_member",
        project_ids=["acme_mkt"],
    )

    with pytest.raises(PermissionError):
        secure_marketing_query(
            index,
            [0.1] * 1536,
            ctx,
            project_id="acme_sales",  # ← they don't belong
        )


def test_admin_sees_all_visibilities():
    """An admin sees every visibility level."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="admin",
        project_ids=["acme_mkt", "acme_sales"],
    )

    # Insert an admin_only doc into the acme_mkt project
    # The admin must see it
    response = secure_marketing_query(
        index,
        [0.1] * 1536,
        ctx,
        project_id="acme_mkt",
    )

    has_admin_only = any(
        m["metadata"]["visibility"] == "admin_only"
        for m in response["matches"]
    )
    assert has_admin_only, "The admin didn't see the admin_only docs"


def test_team_member_cannot_see_admin_only():
    """A team_member does NOT see admin_only docs."""
    ctx = MarketingTenantContext(
        workspace_id="acme",
        user_id="u1",
        user_role="team_member",
        project_ids=["acme_mkt"],
    )

    response = secure_marketing_query(
        index,
        [0.1] * 1536,
        ctx,
        project_id="acme_mkt",
    )

    for match in response["matches"]:
        assert match["metadata"]["visibility"] != "admin_only", (
            f"LEAK: a team_member saw the admin_only doc {match['id']}"
        )


# Wire it into CI
# .github/workflows/ci.yml
# - run: pytest tests/test_marketing_isolation.py

# Every query logs:
{
    "timestamp": "2026-05-15T...",
    "workspace_id": tenant.workspace_id,
    "user_id": tenant.user_id,
    "user_role": tenant.user_role,
    "namespace_used": namespace,
    "project_id": project_id,
    "query_hash": hash(query),  # NOT the plaintext query
    "results_count": len(response["matches"]),
}

# 7-year retention for SOC2
# Storage: S3 with object lock (write-once)

Summary and next step

What you learned:

• Pinecone's namespaces are structural isolation — the engine guarantees the separation, not the code.
• Better than a metadata filter for multi-tenancy: even if the filter fails, the namespace prevents the leak.
• Combine them: a namespace for the tenant, a metadata filter for sub-segmentation (visibility, project_id, date).
• A single tenant_namespace() function prevents inconsistency across devs.
• A mandatory secure_query_pinecone wrapper that enforces the right namespace.
• Automated isolation tests are critical for production.
• The traps: forgetting the namespace in a query (the default), inconsistent naming, slow cross-namespace queries, namespaces with invalid characters.

Checkpoint: before moving on, you should be able to:

• Design tenant_namespace() with sanitization.
• Implement secure_query_pinecone with the namespace forced in.
• Write automated isolation tests that run in CI.

Next capsule: 06 — Metadata filtering in Pinecone.

You have namespaces for the tenant. Now we cover Pinecone's native metadata filtering — slightly different syntax from ChromaDB, the same concepts, better performance. You'll learn to migrate your M06 filters to Pinecone correctly.

Resources

1. Pinecone — Multitenancy Guide — The official pattern
2. Pinecone — Namespaces — The feature's documentation
3. OWASP API Security — The threat model
4. Microsoft — Multi-tenant SaaS Patterns — The architectural decision
5. SOC2 Trust Services Criteria — Compliance
6. Anthropic — Contextual Retrieval — A complementary pattern

Estimated time: 30-35 minutes Next: 06-metadata-filtering-in-pinecone.md