API reference#

The core is importable from needtoknow and has no dependencies. Adapters live in their own modules and need their extra.

Retrieval#

class needtoknow.PermissionedRetriever(backend, authorizer, *, strategy='auto', action='read', overfetch=4, max_candidates=1000, recheck=True, observers=(), audit_key=None, reranker=None, rerank_depth=50)[source]#

Searches a backend and returns only the hits the subject may read.

strategy chooses where permissions are applied: "prefilter" inside the search, with a filter from the authorizer; "postfilter" after it, checking each candidate; "auto" (the default) pre-filters when the backend supports filters and the authorizer returns one. action is passed to the authorizer on every call.

Post-filtering fetches k * overfetch candidates and multiplies the fetch by overfetch until k permitted hits are found, the backend has no more, or max_candidates is reached. recheck checks pre-filtered hits again with the authorizer, so a backend that ignores the filter cannot leak; with recheck=False they are matched against the filter locally instead, which needs the backend to return each chunk’s metadata as stored (the conformance kit checks that). observers receive a RetrievalEvent for every call, and audit_key adds an HMAC of the query to those events.

reranker reorders the result after the permission check: the retriever finds the top rerank_depth hits the subject may read (or k, if larger), the reranker orders them, and the first k are returned. The reranker never sees a chunk the subject may not read.

Parameters:

backend (SearchBackend)
authorizer (Authorizer)
strategy (Strategy)
action (str)
overfetch (int)
max_candidates (int)
recheck (bool)
observers (Sequence[Observer])
audit_key (bytes | None)
reranker (Reranker | RerankFunction | None)
rerank_depth (int)

retrieve(query, subject, k=5)[source]#

Return at most k hits the subject may read, best first.

query is text, or a Query carrying a vector you computed. Raises AuthorizationError when the authorizer fails or gives a malformed answer, and ValueError or TypeError for invalid arguments. Nothing is returned in any of these cases.

Parameters:

query (str | Query)
subject (Subject)
k (int)

Return type:

RetrievalResult

async aretrieve(query, subject, k=5)[source]#

The async form of retrieve, with the same arguments and guarantees.

It uses the backend’s asearch and the authorizer’s afilter_for and acheck when they exist, and runs the sync methods in a thread otherwise.

Parameters:

query (str | Query)
subject (Subject)
k (int)

Return type:

RetrievalResult

class needtoknow.RetrievalResult(hits, strategy, examined, denied, violations, complete)[source]#

What one retrieval returned, with the counts needed to monitor it.

Parameters:

hits (list[Hit])
strategy (Literal['prefilter', 'postfilter'])
examined (int)
denied (int)
violations (int)
complete (bool)

hits: list[Hit]#: Permitted hits, best first, at most k.

strategy: Literal['prefilter', 'postfilter']#: The strategy that was actually used.

examined: int#: Candidates the authorizer checked.

denied: int#: Post-filtered candidates the subject may not read.

violations: int#: Pre-filtered hits that failed the re-check. Should always be 0; anything else means the backend did not apply the filter and deserves an alert.

complete: bool#: False when needtoknow knows results may be missing: post-filtering stopped at max_candidates, or the re-check removed pre-filtered hits. It trusts the backend’s ranking, so it cannot see results an approximate index failed to return.

class needtoknow.SearchBackend(*args, **kwargs)[source]#

A vector store or search index. Return hits ordered best first.

supports_filters says whether search honours the filter argument. Backends may also provide async def asearch(...) with the same signature; otherwise search runs in a thread.

query is what the caller passed to retrieve, unchanged: a string, or a Query when the caller supplies a vector. Query.of(query) gives a Query either way. A backend that takes only strings still works with every string query.

class needtoknow.Reranker(*args, **kwargs)[source]#

Reorders the hits a subject may read: a cross-encoder, a hosted rerank API, any model.

rerank receives the query as it was given to retrieve and the permitted hits, best first, and returns some or all of those hits in its order, with its scores. It may also provide async def arerank(...) with the same signature. A plain function with the same signature can be passed instead.

class needtoknow.RankFusion(*backends, weights=None, rank_constant=60.0, depth=None)[source]#

A backend that searches several backends and fuses their rankings (reciprocal rank fusion).

weights (one per backend, default 1 each) scale each backend’s contribution. rank_constant (default 60, as in the original RRF paper) damps the advantage of the very top ranks. depth is how many results each backend is asked for (default: the k requested). Filters are supported when every backend supports them. Every backend gets the same query: a Query with both text and vector gives keyword backends the text and vector backends the vector, so the query is embedded once, by you, however many vector backends there are.

Parameters:

backends (SearchBackend)
weights (Sequence[float] | None)
rank_constant (float)
depth (int | None)

Strategy is Literal["auto", "prefilter", "postfilter"].

Subjects, documents, hits and queries#

class needtoknow.Subject(id, groups=(), attributes=None)[source]#

Who is asking: a user id, the groups they belong to, and any attributes your policy needs.

Parameters:

id (str)
groups (frozenset[str])
attributes (Mapping[str, Any])

id: str#

groups: frozenset[str] = frozenset({})#

attributes: Mapping[str, Any]#

class needtoknow.Document(id, text, metadata=<factory>)[source]#

A chunk as it is stored: an id, its text, and metadata that may carry permissions.

Parameters:

id (str)
text (str)
metadata (Mapping[str, Any])

id: str#

text: str#

metadata: Mapping[str, Any]#

class needtoknow.Hit(id, text, score, metadata=<factory>)[source]#

A search result: the chunk plus the backend’s relevance score (higher is better).

Parameters:

id (str)
text (str)
score (float)
metadata (Mapping[str, Any])

id: str#

text: str#

score: float#

metadata: Mapping[str, Any]#

class needtoknow.Query(text=None, vector=None)[source]#

What to search for: text, a vector you computed, or both.

A plain string is a text query. Pass a Query with a vector to search with an embedding you made yourself: an image’s, one your embedding service returned, or one computed once and shared by several backends. Vector search uses vector when it is given and embeds text otherwise; keyword search uses text. Give both to fuse vector and keyword results with RankFusion.

Parameters:

text (str | None)
vector (tuple[float, ...] | None)

classmethod of(query)[source]#

query as a Query: a string becomes a text query.

Parameters:: query (str | Query)
Return type:: Query

embedded(embed)[source]#

The vector to search with: vector if given, otherwise embed(text).

Parameters:: embed (Callable[[str], Sequence[float]] | None)
Return type:: list[float]

keywords()[source]#

The text to match words against; keyword search cannot use a vector.

Return type:: str

Filters#

Filter is In | And | Or.

class needtoknow.In(field, values)[source]#

True when the metadata field equals one of values.

For list-valued fields (for example allowed_groups: ["sales", "hr"]) it is true when the field and values share at least one element. A missing field never matches, and neither do metadata values that are not strings or integers.

Parameters:

field (str)
values (frozenset[str | int])

class needtoknow.And(*clauses)[source]#

True when every clause is true.

Parameters:: clauses (tuple[In | And | Or, ...])

class needtoknow.Or(*clauses)[source]#

True when at least one clause is true.

Parameters:: clauses (tuple[In | And | Or, ...])

needtoknow.matches(f, metadata)[source]#

Evaluate a filter against one chunk’s metadata.

Parameters:

f (In | And | Or)
metadata (Mapping[str, Any])

Return type:

bool

needtoknow.filter_to_dict(f)[source]#

JSON-friendly form: {“in”: {“field”: …, “values”: […]}}, {“and”: […]}, {“or”: […]}.

Parameters:: f (In | And | Or)
Return type:: dict[str, Any]

needtoknow.filter_from_dict(data)[source]#

Parse the form produced by filter_to_dict. Anything unexpected raises ValueError, including filters nested more than 100 levels deep.

Parameters:: data (Any)
Return type:: In | And | Or

Authorizers#

class needtoknow.Authorizer(*args, **kwargs)[source]#

check(subject, hits, action='read')[source]#

One decision per hit, in the same order.

Parameters:

subject (Subject)
hits (Sequence[Hit])
action (str)

Return type:

list[bool]

filter_for(subject, action='read')[source]#

A filter matching exactly the chunks the subject may access, or None.

Parameters:

subject (Subject)
action (str)

Return type:

In | And | Or | None

exception needtoknow.AuthorizationError[source]#: The authorizer could not make a decision. No results are returned.

class needtoknow.MetadataACLAuthorizer(*, users_field='allowed_users', groups_field='allowed_groups', everyone='*', actions=('read',))[source]#

Permissions stored on each chunk, in two metadata fields.

allowed_users lists user ids and allowed_groups lists group names. A chunk whose allowed_groups contains the everyone marker (default "*") is readable by every subject. Chunks without either field are readable by nobody: missing permissions mean no access.

This is the simplest model and needs no external service. It suits documents whose permissions are copied onto the chunks at ingestion time.

Parameters:

users_field (str)
groups_field (str)
everyone (str)
actions (Sequence[str])

class needtoknow.FunctionAuthorizer(check, filter_for=None)[source]#

Adapts your own permission logic.

check(subject, hits, action) -> list[bool] is required. Pass filter_for as well if your rules can be turned into a filter; otherwise results are post-filtered.

Parameters:

check (Callable[[Subject, Sequence[Hit], str], Sequence[bool]])
filter_for (Callable[[Subject, str], Filter | None] | None)

class needtoknow.TenantAuthorizer(field='tenant', attribute=None)[source]#

Chunks are readable only within the subject’s tenant.

A chunk is readable when its metadata field field holds the subject’s attribute attribute (both "tenant" by default). A subject without that attribute raises AuthorizationError, so a request that lost its tenant returns nothing rather than everything. Combine it with fine-grained permissions: AllOf(TenantAuthorizer(), OpenFGAAuthorizer(...)).

Parameters:

field (str)
attribute (str | None)

class needtoknow.AllOf(*authorizers)[source]#

Allows a chunk only when every authorizer allows it.

For example tenant isolation plus fine-grained permissions: AllOf(TenantAuthorizer(), OpenFGAAuthorizer(...)). Authorizers are asked in order, each only about the chunks every earlier one allowed, so put cheap ones first. The pre-filter is the conjunction of every authorizer’s filter, or none when one of them has none; scope_for then gives the conjunction of the filters that do exist, so post-filtering searches only inside them (for example only the subject’s tenant).

Parameters:: authorizers (Authorizer)

class needtoknow.AnyOf(*authorizers)[source]#

Allows a chunk when at least one authorizer allows it.

For example documents shared directly or through a policy: AnyOf(MetadataACLAuthorizer(), OPAAuthorizer(...)). Authorizers are asked in order, each only about the chunks every earlier one denied. The pre-filter is the disjunction of every authorizer’s filter, or none when one of them has none; scope_for is the disjunction of every authorizer’s filter or scope, or none when one has neither.

Parameters:: authorizers (Authorizer)

class needtoknow.CachingAuthorizer(authorizer, *, ttl=30.0, maxsize=100000, shared=None, namespace='needtoknow', local_ttl=1.0, clock=<built-in function monotonic>)[source]#

Caches another authorizer’s answers per subject and action for ttl seconds.

ttl is the revocation bound: a permission revoked at the source stops granting access at most ttl seconds later. maxsize caps the number of entries in this process, dropping the least recently used first. shared adds a cache all processes share (for example RedisCache), under keys starting with namespace; this process then keeps its own copy of each answer for at most local_ttl seconds, which bounds how long an invalidate() made in another process takes to reach this one. Errors and malformed answers are never cached. clock is for tests.

Every CachingAuthorizer with the same shared cache and namespace uses the same answers, whatever authorizer it wraps: give each differently configured authorizer (another policy, store, tenant model or action mapping) a namespace of its own.

Parameters:

authorizer (Authorizer)
ttl (float)
maxsize (int)
shared (SharedCache | None)
namespace (str)
local_ttl (float)
clock (Callable[[], float])

stats()[source]#

What this cache has done since it was created, and how many answers it holds now.

Return type:: CacheStats

invalidate(subject=None)[source]#

Drop every cached decision, or those for one subject id, here and in the shared cache.

subject is a Subject or its id; the answers for every subject with that id are dropped, whatever groups and attributes they came with. An answer the authorizer gave before this call, to a retrieval still in progress, is not cached afterwards, here or in the shared cache. Other processes stop using their own copies within local_ttl seconds. A failure to reach the shared cache is raised: the caller asked for the revocation now.

Parameters:: subject (Subject | str | None)
Return type:: None

class needtoknow.caching.CacheStats(hits, shared_hits, misses, shared_failures, entries)[source]#

What a CachingAuthorizer has done since it was created, from stats(), for monitoring.

Parameters:

hits (int)
shared_hits (int)
misses (int)
shared_failures (int)
entries (int)

hits: int#: Answers served from this process’s own cache.

shared_hits: int#: Answers served from the shared cache.

misses: int#: Answers the wrapped authorizer was asked for.

shared_failures: int#: Reads and writes of the shared cache that failed; each failed read was treated as a miss.

entries: int#: Answers this process holds now.

class needtoknow.caching.SharedCache(*args, **kwargs)[source]#

A cache every process shares, such as Redis. Keys are strings and values opaque bytes.

get_many(keys)[source]#

The value of each key, or None when it is absent.

Parameters:: keys (Sequence[str])
Return type:: list[bytes | None]

set_many(entries, ttl)[source]#

Store each entry for ttl seconds.

Parameters:

entries (Mapping[str, bytes])
ttl (float)

Return type:

None

delete_prefix(prefix)[source]#

Delete every key that starts with prefix.

Parameters:: prefix (str)
Return type:: None

class needtoknow.caching.RedisCache(client, *, signing_key=None)[source]#

A SharedCache on Redis or a Redis-compatible store (Valkey, Dragonfly), with redis-py.

Install with pip install "needtoknow[redis]" and pass a redis.Redis (or RedisCluster) client. Entries expire in Redis after the cache’s ttl.

Anyone who can write to this Redis could otherwise plant a decision. Pass signing_key (random bytes known only to your application, for example 32 from secrets.token_bytes) and every value is signed with HMAC-SHA256 over its key and content: an entry without a valid signature is ignored, so the authorizer is asked instead.

Parameters:

client (Any)
signing_key (bytes | None)

class needtoknow.authorizers.opa.OPAAuthorizer(url, *, decisions_path='needtoknow/decisions', filter_path=None, timeout=2.0, headers=None, retries=1, circuit_breaker=None, ssl_context=None)[source]#

Asks an Open Policy Agent server for decisions over its HTTP API.

decisions_path is the rule returning one boolean per resource and filter_path, if set, the rule returning a filter or null. headers are sent with every request, for example an Authorization header. retries is how many times a transient failure is retried, circuit_breaker pauses requests to a server that keeps failing, and ssl_context sets the TLS trust and client certificates (your own CA, mutual TLS). Every failure raises AuthorizationError.

Parameters:

url (str)
decisions_path (str)
filter_path (str | None)
timeout (float)
headers (Mapping[str, str] | None)
retries (int)
circuit_breaker (CircuitBreaker | None)
ssl_context (ssl.SSLContext | None)

class needtoknow.authorizers.openfga.OpenFGAAuthorizer(url, store_id, *, authorization_model_id=None, relations=None, object_type='document', user_type='user', resource_field='doc_id', group_type='group', group_relation='member', contextual_groups=True, consistency=None, list_objects_limit=1000, list_objects_deadline=3.0, batch_size=50, timeout=2.0, headers=None, retries=1, circuit_breaker=None, ssl_context=None)[source]#

Asks an OpenFGA store for decisions over its HTTP API.

relations maps actions to relations (default {"read": "viewer"}). Users are <user_type>:<subject id> and objects <object_type>:<resource id>, where the resource id is the metadata field resource_field (or the chunk id when it is None). consistency is "HIGHER_CONSISTENCY", "MINIMIZE_LATENCY" or None (the server’s default). list_objects_limit and list_objects_deadline are the server’s ListObjects result limit and deadline in seconds (None for a server without a deadline): an answer that reaches either may be partial, so it gives no filter. retries, circuit_breaker and ssl_context work as in OPAAuthorizer. Every failure raises AuthorizationError.

Parameters:

url (str)
store_id (str)
authorization_model_id (str | None)
relations (Mapping[str, str] | None)
object_type (str)
user_type (str)
resource_field (str | None)
group_type (str)
group_relation (str)
contextual_groups (bool)
consistency (str | None)
list_objects_limit (int)
list_objects_deadline (float | None)
batch_size (int)
timeout (float)
headers (Mapping[str, str] | None)
retries (int)
circuit_breaker (CircuitBreaker | None)
ssl_context (ssl.SSLContext | None)

class needtoknow.authorizers.spicedb.SpiceDBAuthorizer(url, token=None, *, permissions=None, resource_type='document', subject_type='user', resource_field='doc_id', group_type='group', group_relation='member', group_subjects=True, consistency=None, caveat_context=None, lookup_limit=1000, batch_size=100, timeout=2.0, headers=None, retries=1, circuit_breaker=None, ssl_context=None)[source]#

Asks SpiceDB for decisions over its HTTP API.

token is SpiceDB’s preshared key (sent as a bearer token). permissions maps actions to SpiceDB permissions (default {"read": "view"}). Resources are <resource_type>:<resource id> and subjects <subject_type>:<subject id>, plus <group_type>:<group>#<group_relation> for each of the subject’s groups when group_subjects is true. caveat_context(subject) returns the context for caveated relationships. retries, circuit_breaker and ssl_context work as in OPAAuthorizer. Every failure raises AuthorizationError.

Parameters:

url (str)
token (str | None)
permissions (Mapping[str, str] | None)
resource_type (str)
subject_type (str)
resource_field (str | None)
group_type (str)
group_relation (str)
group_subjects (bool)
consistency (Mapping[str, Any] | None)
caveat_context (Callable[[Subject], Mapping[str, Any]] | None)
lookup_limit (int)
batch_size (int)
timeout (float)
headers (Mapping[str, str] | None)
retries (int)
circuit_breaker (CircuitBreaker | None)
ssl_context (ssl.SSLContext | None)

class needtoknow.authorizers.cerbos.CerbosAuthorizer(url, *, actions=None, resource_kind='document', resource_field=None, roles=('user',), policy_version=None, scope=None, batch_size=50, timeout=2.0, headers=None, retries=1, circuit_breaker=None, ssl_context=None)[source]#

Asks a Cerbos PDP for decisions over its HTTP API.

actions maps needtoknow actions to Cerbos actions (default {"read": "read"}). roles is a list of roles for every principal, or a function of the subject (default ["user"]). Resources are of kind resource_kind, with the chunk id as id (or resource_field from the metadata) and the chunk’s metadata as attributes. policy_version and scope are passed to both principal and resources. retries, circuit_breaker and ssl_context work as in OPAAuthorizer. Every failure raises AuthorizationError.

Parameters:

url (str)
actions (Mapping[str, str] | None)
resource_kind (str)
resource_field (str | None)
roles (Sequence[str] | Callable[[Subject], Sequence[str]])
policy_version (str | None)
scope (str | None)
batch_size (int)
timeout (float)
headers (Mapping[str, str] | None)
retries (int)
circuit_breaker (CircuitBreaker | None)
ssl_context (ssl.SSLContext | None)

needtoknow.authorizers.cerbos.plan_to_filter(plan)[source]#

Translate a PlanResources filter into a needtoknow filter, or None when it cannot be expressed exactly (the retriever then post-filters).

Parameters:: plan (Mapping[str, Any])
Return type:: In | And | Or | None

class needtoknow.authorizers.cedar.CedarAuthorizer(policies, *, entities=(), actions=None, principal_type='User', group_type='Group', resource_type='Document', schema=None, batch_size=100)[source]#

Evaluates Cedar policies in this process, with cedarpy.

policies is the policy text. entities adds entities to every request, such as a group hierarchy or the folders documents belong to: a list of Cedar entity JSON objects, or a function of the subject returning them. An entity for one of the subject’s groups, with its parent groups say, replaces the empty one needtoknow adds; entities must not repeat the principal or the chunks. actions maps needtoknow actions to Cedar action ids (default {"read": "read"}, so Action::"read"). schema, when given, is passed to Cedar, which then validates requests and entities against it. batch_size is how many chunks one Cedar call decides.

Parameters:

policies (str)
entities (Sequence[Entity] | Callable[[Subject], Sequence[Entity]])
actions (Mapping[str, str] | None)
principal_type (str)
group_type (str)
resource_type (str)
schema (str | Mapping[str, Any] | None)
batch_size (int)

class needtoknow.authorizers.CircuitBreaker(failures=5, reset_after=30.0, *, clock=<built-in function monotonic>)[source]#

Pauses requests to a permission service that keeps failing.

After failures consecutive failed requests the breaker opens, and every call raises AuthorizationError at once, without contacting the service, for reset_after seconds. Then one request is let through as a probe: if it succeeds the breaker closes, if it fails the breaker opens again. A failure is a request that got no HTTP response, or a 5xx one; any other response shows that the service is up. Retrieval fails closed while the breaker is open, as it would if every request failed, but a struggling service is not flooded with requests.

One breaker can be shared by several authorizers that call the same service.

Parameters:

failures (int)
reset_after (float)
clock (Callable[[], float])

property is_open: bool#: True while requests are paused, for monitoring.

Permission sync#

See Google Drive.

class needtoknow.sync.Access(users=(), groups=(), everyone=False)[source]#

Who may read a document: these users, the members of these groups, or everyone.

Access() grants nobody: the safe answer when a source says nothing.

Parameters:

users (frozenset[str])
groups (frozenset[str])
everyone (bool)

metadata(*, users_field='allowed_users', groups_field='allowed_groups', everyone='*')[source]#

The permission fields for MetadataACLAuthorizer, with its default field names and marker.

Both fields are always present, so writing them replaces every earlier grant. A group named like the everyone marker raises ValueError: stored, it would let everyone read.

Parameters:

users_field (str)
groups_field (str)
everyone (str)

Return type:

dict[str, list[str]]

needtoknow.sync.with_access(metadata, access, *, users_field='allowed_users', groups_field='allowed_groups', everyone='*')[source]#

A chunk’s metadata with its permission fields replaced by access, and every other key kept.

Parameters:

metadata (Mapping[str, Any])
access (Access)
users_field (str)
groups_field (str)
everyone (str)

Return type:

dict[str, Any]

class needtoknow.sync.google_drive.GoogleDriveACL(service, *, link_sharing=False, domain_prefix='domain:', domains=None)[source]#

Reads who may read Drive files, through a Drive v3 service.

service is googleapiclient.discovery.build("drive", "v3", credentials=...). Its credentials need the drive.readonly scope (shared drives’ settings need more than drive.metadata.readonly) and must see each file’s sharing and each shared drive’s settings.

domains lists your organization’s domains. When given, grants outside them count for nothing. A file in a shared drive that is open only to its domain’s users needs them: reading its access without them raises ValueError. Errors from the API, such as a file the credentials cannot see, are raised too: a document whose permissions could not be read should keep the permissions it has, not lose or gain any.

Parameters:

service (Any)
link_sharing (bool)
domain_prefix (str)
domains (Iterable[str] | None)

permissions(file_id)[source]#

Every permission on the file, across all pages.

Parameters:: file_id (str)
Return type:: list[Mapping[str, Any]]

access(file_id)[source]#

Who may read the file, within its shared drive’s restrictions.

Parameters:: file_id (str)
Return type:: Access

needtoknow.sync.google_drive.drive_access(permissions, *, link_sharing=False, domain_prefix='domain:', members_only=False, domains=None, now=None)[source]#

Who may read a file, from the permission resources permissions.list returns for it.

Ask for the fields in FIELDS. link_sharing=True counts sharing by link as access for the whole domain or for everyone. A domain grant gives the group domain_prefix + the domain. members_only=True counts only grants to a shared drive’s members. When domains is given, only grants inside those domains count, and an “anyone” grant gives each of their domain groups instead of everyone. Expiry times are compared with now, the current time by default.

Parameters:

permissions (Iterable[Mapping[str, Any]])
link_sharing (bool)
domain_prefix (str)
members_only (bool)
domains (Iterable[str] | None)
now (datetime | None)

Return type:

Access

needtoknow.sync.google_drive.FIELDS = 'nextPageToken,permissions(id,type,role,emailAddress,domain,allowFileDiscovery,deleted,expirationTime,view,permissionDetails,inheritedPermissionsDisabled)'#: The fields GoogleDriveACL asks Drive for: everything drive_access reads.

needtoknow.sync.google_drive.READ_ROLES = frozenset({'commenter', 'fileOrganizer', 'organizer', 'owner', 'reader', 'writer'})#: Every Drive role that can read a file’s content.

Events, audit and tracing#

class needtoknow.RetrievalEvent(subject_id, action, k, strategy, returned_ids=(), examined=0, denied=0, violations=0, complete=False, duration_ms=0.0, error=None, query_hmac=None, started_at=0.0, search_ms=0.0, authorize_ms=0.0, searches=0, rerank_ms=0.0, reranked=0, cause=None)[source]#

One retrieval, for audit logs and monitoring. Never contains query or chunk text.

Parameters:

subject_id (str)
action (str)
k (int)
strategy (str | None)
returned_ids (tuple[str, ...])
examined (int)
denied (int)
violations (int)
complete (bool)
duration_ms (float)
error (str | None)
query_hmac (str | None)
started_at (float)
search_ms (float)
authorize_ms (float)
searches (int)
rerank_ms (float)
reranked (int)
cause (str | None)

subject_id: str#

action: str#

k: int#

strategy: str | None#: The strategy that ran, or None if the call failed before choosing one.

returned_ids: tuple[str, ...] = ()#

examined: int = 0#

denied: int = 0#

violations: int = 0#

complete: bool = False#

duration_ms: float = 0.0#

error: str | None = None#: The exception type when the call failed, for example “AuthorizationError”.

query_hmac: str | None = None#

started_at: float = 0.0#: Unix time when the call started.

search_ms: float = 0.0#: Time spent in the backend, over every search of this call.

authorize_ms: float = 0.0#: Time spent in the authorizer: filters, scopes and checks.

searches: int = 0#: Backend searches made: one when pre-filtering, more when post-filtering fetched again.

rerank_ms: float = 0.0#: Time spent in the reranker.

reranked: int = 0#: Permitted chunks given to the reranker; 0 when there is none.

cause: str | None = None#: The type of the exception behind error, when there is one: for an AuthorizationError, what the call to the permission service raised, such as “TimeoutError” or “ConnectionRefusedError”. None when the error has no cause, as for a malformed answer.

to_dict()[source]#

Return type:: dict[str, Any]

class needtoknow.JsonLinesAudit(stream)[source]#

Writes one JSON object per retrieval to a text stream (a file, stdout, a socket wrapper).

It records who asked, what came back and what was denied, never the query or chunk text.

Parameters:: stream (IO[str])

needtoknow.events.query_fingerprint(query, key)[source]#

HMAC-SHA256 of the query under key, in hex. Equal queries give equal fingerprints.

A Query with text is fingerprinted by its text, so the same question matches whether or not a vector came with it. A vector-only query is fingerprinted by its vector.

Parameters:

query (str | Query)
key (bytes)

Return type:

str

class needtoknow.otel.OpenTelemetryObserver(tracer=None, *, record_subject=False)[source]#

Records each RetrievalEvent as a span named needtoknow.retrieve.

tracer defaults to the needtoknow tracer of the global tracer provider. Spans carry counts, outcomes and the time spent in the backend and the authorizer, never query or chunk text; the subject id only with record_subject=True.

Parameters:

tracer (Any)
record_subject (bool)

class needtoknow.otel.OpenTelemetryMetrics(meter=None)[source]#

Records each RetrievalEvent as OpenTelemetry metrics.

meter defaults to the needtoknow meter of the global meter provider. Every instrument is broken down by needtoknow.action and needtoknow.strategy only, plus the outcome and error.type where they apply. These have a handful of values each, so the metrics stay cheap to store; subjects, queries and chunks are never recorded.

gen_ai.client.operation.duration (histogram, s): each retrieval, with gen_ai.operation.name = "retrieval".
needtoknow.retrievals (counter): retrievals by needtoknow.outcome, which is complete, incomplete or error.
needtoknow.search.duration, needtoknow.authorize.duration (histograms, s): time spent in the backend and in the authorizer during each retrieval.
needtoknow.rerank.duration (histogram, s): time spent in the reranker, for retrievals that reranked.
needtoknow.chunks.returned, needtoknow.chunks.examined, needtoknow.chunks.denied (counters): chunks returned, checked by the authorizer, and withheld.
needtoknow.violations (counter): pre-filtered chunks the re-check rejected. Alert on any increase.

Parameters:: meter (Any)

An observer is any callable that takes a RetrievalEvent.

Backends#

class needtoknow.InMemoryBackend(documents=())[source]#

Keeps documents in memory and ranks them by keyword overlap.

For tests, examples and small corpora. Its filtering follows matches exactly, which makes it the reference the other backends are tested against.

Parameters:: documents (Iterable[Document])

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

class needtoknow.backends.pgvector.PgVectorBackend(pool, embed=None, *, table='needtoknow_chunks', iterative_scan='relaxed_order', ef_search=None, mode='vector', text_search='english', async_pool=None)[source]#

Cosine-similarity search over a pgvector table, with the filter in the WHERE clause.

pool is a psycopg ConnectionPool; embed turns text into a vector (your embedding model). Without embed, search with Query(vector=...) and pass embeddings to add. Scores are cosine similarity, 1 - cosine distance. Pass async_pool (an AsyncConnectionPool) for a native asearch; otherwise async calls run search in a thread. ef_search sets pgvector’s hnsw.ef_search (1 to 1000) for each search: higher finds more of the true nearest neighbours and takes longer. None leaves the server’s setting (40 by default).

mode="keyword" searches the text with PostgreSQL full-text search instead (websearch_to_tsquery in the text_search configuration, ranked by ts_rank_cd), under the same permission condition, for hybrid search with RankFusion; embed may then be None. Create the matching index with create_schema(..., text_search=...).

Parameters:

pool (Any)
embed (Embed | None)
table (str)
iterative_scan (str | None)
ef_search (int | None)
mode (str)
text_search (str)
async_pool (Any)

add(documents, embeddings=None)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.pgvector.create_schema(pool, dim, *, table='needtoknow_chunks', hnsw=True, text_search=None)[source]#

Create the extension, table and indexes if they do not exist. pool is a psycopg ConnectionPool.

text_search (a text search configuration, such as "english") also creates the full-text GIN index that PgVectorBackend(mode="keyword", text_search=...) uses; pass the same name to both.

Parameters:

pool (Any)
dim (int)
table (str)
hnsw (bool)
text_search (str | None)

Return type:

None

needtoknow.backends.pgvector.filter_to_sql(f)[source]#

Translate a filter into a condition on the metadata jsonb column, with bound parameters.

Each value becomes two containment tests: scalar equality ({"field": v}) and array membership ({"field": [v]}), which the GIN index on metadata serves. Above MANY_VALUES values (a subject in thousands of groups) the test takes two array parameters instead of two per value, so the query stays small and far below PostgreSQL’s limit of 65,535 parameters. jsonb equality and containment compare by type either way, matching the reference semantics in needtoknow.filters.matches.

Parameters:: f (In | And | Or | None)
Return type:: tuple[str, list[Any]]

class needtoknow.backends.chroma.ChromaBackend(collection, embed=None)[source]#

Search a Chroma collection with the permission filter applied by Chroma.

If embed is given, queries and added documents are embedded with it; otherwise the collection’s own embedding function is used. A Query with a vector is searched with that vector. Scores are 1 - distance, so create the collection with cosine distance for scores in [-1, 1].

Parameters:

collection (Any)
embed (Embed | None)

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.chroma.filter_to_where(f)[source]#

Translate a filter into a Chroma where document, or NEVER when nothing can match.

In becomes $in on the scalar field (one clause per value type, since Chroma’s $in lists are single-typed) plus an equality test on each value’s list key.

Parameters:: f (In | And | Or)
Return type:: dict[str, Any]

class needtoknow.backends.elasticsearch.ElasticsearchBackend(client, index, embed=None, *, vector_field='embedding', content_field='content', metadata_field='metadata', num_candidates=100, refresh='wait_for', mode='vector', async_client=None)[source]#

Search an Elasticsearch index with the permission filter applied during the kNN search.

embed turns text into a vector; without it, search with Query(vector=...) and pass embeddings to add. num_candidates is how many nearest neighbours each shard considers (at least k, at most 10,000): higher finds more of the true nearest chunks and takes longer. Scores are Elasticsearch’s: (1 + cosine) / 2 for a cosine field. refresh is passed to bulk writes ("wait_for" by default, so written chunks are searchable when add returns). mode="keyword" searches the text with BM25 instead, under the same permission filter, for hybrid search with RankFusion; embed may then be None.

Parameters:

client (Any)
index (str)
embed (Embed | None)
vector_field (str)
content_field (str)
metadata_field (str)
num_candidates (int)
refresh (bool | str)
mode (str)
async_client (Any)

add(documents, embeddings=None)#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.elasticsearch.create_index(client, index, dim, *, index_options=None, settings=None, vector_field='embedding', content_field='content', metadata_field='metadata')[source]#

Create the index if it does not exist, with a cosine dense_vector field.

Elasticsearch quantizes dense vectors by default (int8_hnsw below 384 dimensions, bbq_hnsw from 384 in 9.1 and later), which saves memory and costs some recall. Pass index_options={"type": "hnsw"} to keep full-precision vectors.

Parameters:

client (Any)
index (str)
dim (int)
index_options (Mapping[str, Any] | None)
settings (Mapping[str, Any] | None)
vector_field (str)
content_field (str)
metadata_field (str)

Return type:

None

needtoknow.backends.elasticsearch.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata.

Parameters:: metadata (Mapping[str, Any])
Return type:: list[str]

class needtoknow.backends.opensearch.OpenSearchBackend(client, index, embed=None, *, vector_field='embedding', content_field='content', metadata_field='metadata', num_candidates=100, refresh='wait_for', mode='vector', async_client=None)[source]#

Search an OpenSearch index with the permission filter applied during the k-NN search.

embed turns text into a vector; without it, search with Query(vector=...) and pass embeddings to add. Scores are OpenSearch’s: (1 + cosine) / 2 for a cosinesimil field. refresh is passed to bulk writes ("wait_for" by default, so written chunks are searchable when add returns). num_candidates is not used by OpenSearch’s knn query; set ef_search in the index settings instead. mode="keyword" searches the text with BM25 instead, under the same permission filter, for hybrid search with RankFusion; embed may then be None.

Parameters:

client (Any)
index (str)
embed (Embed | None)
vector_field (str)
content_field (str)
metadata_field (str)
num_candidates (int)
refresh (bool | str)
mode (str)
async_client (Any)

add(documents, embeddings=None)#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.opensearch.create_index(client, index, dim, *, engine='lucene', parameters=None, settings=None, vector_field='embedding', content_field='content', metadata_field='metadata')[source]#

Create the index if it does not exist, with a cosine knn_vector field (HNSW).

engine is "lucene" (the default) or "faiss"; both apply filters during the search. parameters are the HNSW method parameters, for example {"m": 16, "ef_construction": 128}.

Parameters:

client (Any)
index (str)
dim (int)
engine (str)
parameters (Mapping[str, Any] | None)
settings (Mapping[str, Any] | None)
vector_field (str)
content_field (str)
metadata_field (str)

Return type:

None

needtoknow.backends.opensearch.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata.

Parameters:: metadata (Mapping[str, Any])
Return type:: list[str]

class needtoknow.backends.qdrant.QdrantBackend(client, collection, embed=None, *, mode='vector', text_search=False, content_key='page_content', metadata_key='metadata', vector_name=None, search_params=None, async_client=None)[source]#

Search a Qdrant collection with the permission filter applied by Qdrant.

embed turns text into a vector (your embedding model). vector_name selects a named vector, search_params (a qdrant_client.models.SearchParams) tunes the search, for example SearchParams(hnsw_ef=128) for higher recall or SearchParams(exact=True) for brute force. Scores are Qdrant’s: cosine similarity for a cosine collection. Pass async_client (an AsyncQdrantClient) for a native asearch; otherwise async calls run search in a thread. Without embed, search with Query(vector=...) and pass embeddings to add.

mode="keyword" ranks by BM25 instead, on the sparse vector needtoknow_bm25 of a collection created with text_search=True, under the same query_filter. Scores are BM25, higher for better matches. A vector backend writes that sparse vector too when given text_search=True, so one collection serves both halves of hybrid search. A keyword backend without embed writes chunks with only the sparse vector, which keyword search finds and vector search does not.

Parameters:

client (Any)
collection (str)
embed (Embed | None)
mode (str)
text_search (bool)
content_key (str)
metadata_key (str)
vector_name (str | None)
search_params (Any)
async_client (Any)

add(documents, embeddings=None)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.qdrant.create_collection(client, name, dim, *, keyword_fields=('allowed_users', 'allowed_groups'), metadata_key='metadata', vector_name=None, text_search=False)[source]#

Create a cosine-distance collection if it does not exist, and keyword indexes on the permission fields (metadata.allowed_users and metadata.allowed_groups by default).

text_search=True also creates the sparse vector needtoknow_bm25 with Qdrant’s IDF modifier, for mode="keyword". Qdrant can add it to an existing collection only by recreating it.

Parameters:

client (Any)
name (str)
dim (int)
keyword_fields (Sequence[str])
metadata_key (str)
vector_name (str | None)
text_search (bool)

Return type:

None

needtoknow.backends.qdrant.filter_to_qdrant(f, metadata_key='metadata')[source]#

Translate a filter into a Qdrant Filter, or None when nothing can match.

In becomes MatchAny on metadata.<field>, one condition per value type (strings and integers are never mixed in one list), joined with should. An In with no values matches nothing; that is decided here instead of being left to Qdrant’s handling of empty lists.

Parameters:

f (Filter)
metadata_key (str)

Return type:

models.Filter | None

needtoknow.backends.qdrant.point_id(document_id)[source]#

The Qdrant point id for a document id: the id itself if it is a UUID, otherwise a UUID derived from it.

Parameters:: document_id (str)
Return type:: str

needtoknow.backends.qdrant.bm25_terms(text)[source]#

A chunk’s BM25 term weights, as the indices and values of the sparse vector needtoknow_bm25.

Words are runs of letters, digits and underscores, lowercased; every word is kept. Each word’s index is its CRC-32, and its value BM25’s term-frequency part, tf * (k1 + 1) / (tf + k1 * (1 - b + b * length / 256)) with k1 = 1.2 and b = 0.75. Qdrant multiplies in the inverse document frequency at search time. Use it to fill the field for points written by other tools.

Parameters:: text (str)
Return type:: tuple[list[int], list[float]]

class needtoknow.backends.milvus.MilvusBackend(client, collection, embed=None, *, mode='vector', vector_field='embedding', content_field='content', metadata_field='metadata', ef=100, search_params=None, consistency_level='Strong', async_client=None)[source]#

Vector search on Milvus, with the permission filter applied inside the search.

embed turns text into a vector (your embedding model). ef is HNSW’s search breadth: higher finds more of the true nearest neighbours, more slowly; it is raised to k when k is larger, as Milvus requires. search_params replaces the search parameters entirely, for another index type. consistency_level is passed to every search (“Strong” by default, so writes are visible at once; “Bounded” is faster when a short delay is acceptable). Pass async_client (an AsyncMilvusClient) for a native asearch; otherwise async calls run search in a thread. Without embed, search with Query(vector=...) and pass embeddings to add.

mode="keyword" searches the BM25 sparse vector of a collection created with text_search=True instead, under the same permission filter, with the query’s text. Scores are Milvus’s BM25, higher for better matches; search_params is passed only when given. Milvus requires every chunk’s vector, so add still needs embed or embeddings in this mode.

Parameters:

client (Any)
collection (str)
embed (Embed | None)
mode (str)
vector_field (str)
content_field (str)
metadata_field (str)
ef (int)
search_params (dict[str, Any] | None)
consistency_level (str)
async_client (Any)

add(documents, embeddings=None)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.milvus.create_collection(client, name, dim, *, metric='COSINE', vector_field='embedding', content_field='content', metadata_field='metadata', max_id_length=512, max_content_length=65535, max_terms=4096, max_term_length=1024, hnsw=None, text_search=False)[source]#

Create a collection MilvusBackend can use, with an HNSW index on the vector and an inverted index on the permission terms.

max_terms caps the tagged terms per chunk: one per string or integer in its metadata, so a chunk shared with many groups needs more. max_term_length caps one term, which is the field name, two characters and the value.

text_search=True analyzes the text with Milvus’s standard analyzer, which keeps every word, and adds a BM25 function that fills a sparse vector, needtoknow_sparse, with an inverted index, for mode="keyword". Milvus can only add it when the collection is created.

Parameters:

client (Any)
name (str)
dim (int)
metric (str)
vector_field (str)
content_field (str)
metadata_field (str)
max_id_length (int)
max_content_length (int)
max_terms (int)
max_term_length (int)
hnsw (dict[str, int] | None)
text_search (bool)

Return type:

None

needtoknow.backends.milvus.filter_to_expression(f)[source]#

Translate a filter into a Milvus filter template and its parameters.

Returns (expression, params) for search(filter=expression, filter_params=params), or None when nothing can match. Each In becomes ARRAY_CONTAINS_ANY(needtoknow_terms, {pN}) with the tagged terms in params["pN"].

Parameters:: f (In | And | Or)
Return type:: tuple[str, dict[str, list[str]]] | None

needtoknow.backends.milvus.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata, for rows written by other tools.

Parameters:: metadata (Mapping[str, Any])
Return type:: list[str]

class needtoknow.backends.mongodb.MongoDBBackend(collection, embed=None, *, mode='vector', index='needtoknow', text_index=None, vector_field='embedding', content_field='content', metadata_field='metadata', num_candidates=100, exact=False, wait_timeout=60.0, async_collection=None)[source]#

Vector search on MongoDB Atlas Vector Search, with the permission filter applied inside the search.

embed turns text into a vector (your embedding model). index names the vector search index. num_candidates is how many neighbours the approximate search considers (raised to k, at most 10,000): higher finds more of the true nearest chunks and takes longer. exact=True searches every chunk instead. wait_timeout bounds how long add and delete wait for the index. Pass async_collection (the same collection from an AsyncMongoClient) for a native asearch; otherwise async calls run search in a thread. Without embed, search with Query(vector=...) and pass embeddings to add.

mode="keyword" searches the text with Atlas Search (BM25) instead, in the index named text_index ("needtoknow_text" by default; see create_text_index), under the same permission filter. Scores are Atlas Search’s, higher for better matches. Such a backend needs no embed: without one, add writes chunks with no vector, which keyword search still finds. A vector backend given text_index also waits for that index after each write, so writes through it are searchable by both halves of hybrid search when they return.

Parameters:

collection (Any)
embed (Embed | None)
mode (str)
index (str)
text_index (str | None)
vector_field (str)
content_field (str)
metadata_field (str)
num_candidates (int)
exact (bool)
wait_timeout (float)
async_collection (Any)

add(documents, embeddings=None, *, wait=True)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

With wait (the default), returns once the search index shows every chunk as written, or raises TimeoutError after wait_timeout seconds.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)
wait (bool)

Return type:

None

set_metadata(metadata, *, wait=True)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

With wait (the default), returns once the search index shows the new metadata, or raises TimeoutError after wait_timeout seconds.

Parameters:

metadata (Mapping[str, Mapping[str, Any]])
wait (bool)

Return type:

None

delete(ids, *, wait=True)[source]#

Delete chunks by id; with wait, returns once the search index no longer returns them.

Parameters:

ids (Iterable[str])
wait (bool)

Return type:

None

needtoknow.backends.mongodb.create_search_index(collection, dim, *, name='needtoknow', vector_field='embedding', similarity='cosine', wait=True, timeout=120.0)[source]#

Create the vector search index MongoDBBackend uses, creating the collection if needed.

The index holds the vector and, as filter fields, the permission terms, the write marker and _id. With wait, it returns once the index is queryable, or raises TimeoutError.

Parameters:

collection (Any)
dim (int)
name (str)
vector_field (str)
similarity (str)
wait (bool)
timeout (float)

Return type:

None

needtoknow.backends.mongodb.create_text_index(collection, *, name='needtoknow_text', content_field='content', wait=True, timeout=120.0)[source]#

Create the Atlas Search index that mode="keyword" searches, creating the collection if needed.

The index holds the text, with the lucene.standard analyzer, which keeps every word, and as tokens the permission terms, the write marker and _id. With wait, it returns once the index is queryable, or raises TimeoutError.

Parameters:

collection (Any)
name (str)
content_field (str)
wait (bool)
timeout (float)

Return type:

None

needtoknow.backends.mongodb.filter_to_mongo(f)[source]#

Translate a filter into a $vectorSearch filter on the tagged terms, or None when nothing can match.

Parameters:: f (In | And | Or)
Return type:: dict[str, Any] | None

needtoknow.backends.mongodb.filter_to_search(f)[source]#

Translate a filter into an Atlas Search operator on the tagged terms, for $search, or None when nothing can match: in for each In, and compound for And and Or.

Parameters:: f (In | And | Or)
Return type:: dict[str, Any] | None

needtoknow.backends.mongodb.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata, for documents written by other tools.

Parameters:: metadata (Mapping[str, Any])
Return type:: list[str]

class needtoknow.backends.weaviate.WeaviateBackend(client, collection, embed=None, *, vector_name='embedding', content_property='content', metadata_property='metadata_json', mode='vector', async_client=None)[source]#

Vector search on Weaviate, with the permission filter applied inside the search.

embed turns text into a vector (your embedding model). vector_name is the collection’s named vector. Scores are cosine similarity, 1 minus Weaviate’s cosine distance. Pass async_client (a connected WeaviateAsyncClient) for a native asearch; otherwise async calls run search in a thread. Without embed, search with Query(vector=...) and pass embeddings to add.

mode="keyword" searches the text with Weaviate’s BM25 instead, under the same permission filter, for hybrid search with RankFusion; embed may then be None, and scores are BM25’s.

Parameters:

client (Any)
collection (str)
embed (Embed | None)
vector_name (str)
content_property (str)
metadata_property (str)
mode (str)
async_client (Any)

add(documents, embeddings=None)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.weaviate.create_collection(client, name, *, vector_name='embedding', content_property='content', metadata_property='metadata_json')[source]#

Create a collection WeaviateBackend can use: a self-provided cosine HNSW vector, the text, the metadata as unindexed JSON, and the id and permission terms with field tokenization.

Parameters:

client (Any)
name (str)
vector_name (str)
content_property (str)
metadata_property (str)

Return type:

None

needtoknow.backends.weaviate.filter_to_weaviate(f)[source]#

Translate a filter into a Weaviate filter on the tagged terms, or None when nothing can match.

Parameters:: f (In | And | Or)
Return type:: Any

needtoknow.backends.weaviate.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata, for objects written by other tools.

Parameters:: metadata (Mapping[str, Any])
Return type:: list[str]

needtoknow.backends.weaviate.point_id(document_id)[source]#

The Weaviate object id for a document id: the id itself if it is a UUID, otherwise one derived from it.

Parameters:: document_id (str)
Return type:: str

class needtoknow.backends.redis.RedisBackend(client, index, embed=None, *, mode='vector', prefix='needtoknow:', vector_field='embedding', content_field='content', metadata_field='metadata', ef_runtime=None, async_client=None)[source]#

Vector search on the Redis Query Engine, with the permission filter applied inside the search.

embed turns text into a vector (your embedding model). Chunks are hashes under prefix (the document id appended). ef_runtime is HNSW’s search breadth (Redis’s default when None). Scores are cosine similarity, 1 minus Redis’s cosine distance. Pass async_client (a redis.asyncio.Redis) for a native asearch; otherwise async calls run search in a thread. Without embed, search with Query(vector=...) and pass embeddings to add.

mode="keyword" searches the text with BM25 instead, under the same permission filter, in an index created with content_field=. Each word of the query (letters, digits and underscores) becomes a parameter, so none can act as query syntax, and up to 64 distinct words are kept. Scores are Redis’s BM25, higher for better matches. Such a backend needs no embed: without one, add writes chunks with no vector, which keyword search still finds.

Parameters:

client (Any)
index (str)
embed (Embed | None)
mode (str)
prefix (str)
vector_field (str)
content_field (str)
metadata_field (str)
ef_runtime (int | None)
async_client (Any)

add(documents, embeddings=None)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

set_metadata(metadata)[source]#

Replace the metadata of stored chunks, keeping their text and vectors. Ids not stored are ignored.

Parameters:: metadata (Mapping[str, Mapping[str, Any]])
Return type:: None

needtoknow.backends.redis.create_index(client, index, dim, *, prefix='needtoknow:', vector_field='embedding', content_field=None, algorithm='HNSW', m=16, ef_construction=200)[source]#

Create the index if it does not exist: the permission terms as a TAG field and a cosine vector index, over the hashes whose keys start with prefix.

algorithm="HNSW" (the default, tuned by m and ef_construction) is approximate. Redis indexes a hash again whenever any of its fields changes, so set_metadata removes and re-adds the chunk’s vector in the graph, and a small, often-changed graph can then miss a chunk. algorithm="FLAT" compares the query with every vector: exact, and fast enough for indexes of up to some tens of thousands of chunks.

With content_field (the backend’s, "content" by default) the chunks’ text is indexed as well, for mode="keyword", keeping every word: the index has no stop words. An index created without it can add it with FT.ALTER <index> SCHEMA ADD content TEXT.

Parameters:

client (Any)
index (str)
dim (int)
prefix (str)
vector_field (str)
content_field (str | None)
algorithm (str)
m (int)
ef_construction (int)

Return type:

None

needtoknow.backends.redis.filter_to_query(f)[source]#

Translate a filter into a Query Engine expression on the hex-encoded terms, or None when nothing can match.

Parameters:: f (In | And | Or)
Return type:: str | None

needtoknow.backends.redis.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata, for hashes written by other tools.

Parameters:: metadata (Mapping[str, Any])
Return type:: str

class needtoknow.backends.lancedb.LanceDBBackend(table, embed=None, *, mode='vector', vector_field='vector', content_field='content', metadata_field='metadata', nprobes=None, refine_factor=None)[source]#

Vector search on a LanceDB table, with the permission filter applied before ranking.

embed turns text into a vector (your embedding model); without it, search with Query(vector=...) and pass embeddings to add. Scores are cosine similarity, 1 minus LanceDB’s cosine distance. nprobes and refine_factor tune an IVF index if the table has one. Async calls run search in a thread.

mode="keyword" searches the text with the table’s full-text index instead (build it with create_indexes(table, content_field="content")), under the same permission filter. Scores are LanceDB’s BM25, higher for better matches. Such a backend needs no embed: without one, add writes chunks with no vector, which keyword search still finds.

Parameters:

table (Any)
embed (Embed | None)
mode (str)
vector_field (str)
content_field (str)
metadata_field (str)
nprobes (int | None)
refine_factor (int | None)

add(documents, embeddings=None)[source]#

Insert or replace chunks. Embeddings are computed with embed unless given.

Parameters:

documents (Iterable[Document])
embeddings (Iterable[Sequence[float]] | None)

Return type:

None

needtoknow.backends.lancedb.create_table(db, name, dim, *, vector_field='vector', content_field='content', metadata_field='metadata')[source]#

Create the table if it does not exist, and return it. db is a lancedb.connect(...) connection.

Parameters:

db (Any)
name (str)
dim (int)
vector_field (str)
content_field (str)
metadata_field (str)

Return type:

Any

needtoknow.backends.lancedb.create_indexes(table, *, content_field=None)[source]#

Index the permission terms (LABEL_LIST) and the ids, which filters and upserts use. Run it once the table holds data; rows added later are searched correctly before they are indexed.

With content_field (the backend’s, "content" by default) it also builds the full-text index that mode="keyword" searches, keeping every word: no stop words are removed. Rows added later are found by a scan until table.optimize() adds them to the index.

Parameters:

table (Any)
content_field (str | None)

Return type:

None

needtoknow.backends.lancedb.filter_to_sql(f)[source]#

Translate a filter into a where predicate on the hex-encoded terms, or None when nothing can match.

Parameters:: f (In | And | Or)
Return type:: str | None

needtoknow.backends.lancedb.permission_terms(metadata)[source]#

The value of needtoknow_terms for a chunk with this metadata, for rows written by other tools.

Parameters:: metadata (Mapping[str, Any])
Return type:: list[str]

Frameworks#

class needtoknow.langchain.NeedToKnowRetriever(*args, name=None, tags=None, metadata=None, retriever, k=4, subject=None)[source]#

Returns only the chunks the subject may read.

Parameters:

args (Any)
name (str | None)
tags (list[str] | None)
metadata (dict[str, Any] | None)
retriever (PermissionedRetriever)
k (int)
subject (Subject | None)

class needtoknow.langchain.LangChainVectorStoreBackend(vectorstore, *, id_key=None)[source]#

Any LangChain VectorStore as a needtoknow backend (post-filtered).

Documents need an id: Document.id, or the metadata field named by id_key. Scores follow the store’s ranking (best first), since stores disagree on what their raw scores mean. A Query with a vector is searched with similarity_search_by_vector.

Parameters:

vectorstore (Any)
id_key (str | None)

class needtoknow.llamaindex.NeedToKnowRetriever(retriever, subject, *, k=4, **kwargs)[source]#

Returns only the nodes subject may read.

Parameters:

retriever (PermissionedRetriever)
subject (Subject)
k (int)
kwargs (Any)

class needtoknow.llamaindex.PermissionNodePostprocessor(*, callback_manager=<factory>, authorizer, subject, action='read')[source]#

Keeps only the nodes subject may read, as decided by authorizer.

Parameters:

callback_manager (CallbackManager)
authorizer (Any)
subject (Any)
action (str)

class needtoknow.llamaindex.LlamaIndexBackend(index)[source]#

A LlamaIndex index (anything with as_retriever(similarity_top_k=...)) as a backend.

Parameters:: index (Any)

class needtoknow.haystack.NeedToKnowRetriever(*args, **kwargs)[source]#

Returns at most top_k documents the run’s subject may read, best first.

Parameters:

args (Any)
kwargs (Any)

Return type:

Any

run(query, subject, top_k=None, query_embedding=None)[source]#

Retrieve for subject. top_k overrides the component’s default for this run.

query_embedding (from a text embedder, for example) is searched with instead of embedding query again; query may then be empty.

Parameters:

query (str)
subject (Subject)
top_k (int | None)
query_embedding (list[float] | None)

Return type:

dict[str, Any]

async run_async(query, subject, top_k=None, query_embedding=None)[source]#

The async form of run, with the same inputs and outputs.

Parameters:

query (str)
subject (Subject)
top_k (int | None)
query_embedding (list[float] | None)

Return type:

dict[str, Any]

needtoknow.mcp.add_search_tool(server, retriever, subject, *, name='search', description="Search the organisation's documents. Returns only passages the calling user is allowed to read, most relevant first, each with its id, text and relevance score.", limit=5, max_limit=20, metadata=())[source]#

subject is a function from the request’s verified AccessToken to a Subject, or one fixed Subject for a server only that person can reach. The tool takes query and limit (default limit, at most max_limit); metadata names the metadata keys included in each result.

Parameters:

server (MCPServer[Any])
retriever (PermissionedRetriever)
subject (Subject | Callable[[AccessToken], Subject | Awaitable[Subject]])
name (str)
description (str)
limit (int)
max_limit (int)
metadata (Sequence[str])

Return type:

None

Conformance testing#

See Extending needtoknow.

needtoknow.testing.check_backend(factory, *, examples=25, seed=0, max_documents=20, filters_per_example=4, exact=True, query='doc')[source]#

Check a backend against the reference semantics. Raises ConformanceError on the first failure.

factory(documents) must return a new backend holding exactly those documents. For each of examples random corpora (up to max_documents each) it checks that:

a search without a filter returns stored documents only, min(k, stored) of them for k up to the whole corpus, best first, with text and metadata exactly as stored;
a search with a filter (when supports_filters is true) returns only documents matches selects, and min(k, matching) of them, for filters_per_example random nested filters; and asearch, if the backend has one, returns the same ids in the same order;
retrieval through PermissionedRetriever with MetadataACLAuthorizer never returns a document the subject may not read, reports no violations, and returns as many readable documents as were asked for (with "prefilter" and "postfilter" where available);
if the backend has set_metadata, replacing the metadata of some documents (and of an id that is not stored) leaves their text as it was, changes what filters select as matches says it should, keeps no key of the old metadata, and stores nothing new.

exact=False suits approximate indexes, which may miss matching documents: results must still never include a document that does not match, but may contain fewer than expected.

Every search uses query. Pass Query(vector=...) to check a backend searched with vectors you compute, for example Query(QUERY, HashingEmbedder()(QUERY)) against a backend built without embed and filled with add(documents, embeddings=...).

Parameters:

factory (Callable[[Sequence[Document]], SearchBackend])
examples (int)
seed (int)
max_documents (int)
filters_per_example (int)
exact (bool)
query (str | Query)

Return type:

None

needtoknow.testing.check_authorizer(authorizer, subjects=None, hits=None, *, action='read', seed=0)[source]#

Check that an authorizer’s answers are well formed and agree with each other.

Without subjects and hits, random users and groups and allowed_users/allowed_groups metadata are used, which suits authorizers that read permissions from the chunks (MetadataACLAuthorizer, a FunctionAuthorizer over metadata, CachingAuthorizer around them). For an authorizer backed by a service, pass subjects and hits from your test data. For every subject it checks that:

check returns one boolean per hit, and [] for no hits;
the answer for a hit does not depend on the other hits or their order;
when filter_for returns a filter, check agrees with matches(filter, hit.metadata) for every hit. If they differ, pre-filtered and post-filtered results differ, and the re-check of pre-filtered hits reports violations.

Parameters:

authorizer (Authorizer)
subjects (Sequence[Subject] | None)
hits (Sequence[Hit] | None)
action (str)
seed (int)

Return type:

None

exception needtoknow.testing.ConformanceError[source]#: A backend or authorizer disagreed with needtoknow’s reference semantics.

class needtoknow.testing.HashingEmbedder(dim=16)[source]#

Deterministic bag-of-words embeddings for tests: no model, no network.

Texts that share words get similar vectors, so search results are meaningful enough for tests, and the same text always gets the same vector.

Parameters:: dim (int)

needtoknow.testing.random_documents(rng, n)[source]#

n documents containing QUERY, with metadata that mixes strings, integers, lists of either, booleans and missing fields: the cases where backends’ filter translations tend to differ from the reference.

Parameters:

rng (Random)
n (int)

Return type:

list[Document]

needtoknow.testing.random_filter(rng, depth=2)[source]#

A filter over the fields random_documents uses, nested up to depth levels of And/Or.

Parameters:

rng (Random)
depth (int)

Return type:

In | And | Or