Store

class Store(store_name: str, *args: Any, **kwargs: Any)

Bases: Set

Abstract base class for stores.

registry: Final[dict[str, type[Store]]] = {'empty': <class 'kif_lib.store.empty.EmptyStore'>, 'mixer': <class 'kif_lib.store.mixer.MixerStore'>, 'rdf': <class 'kif_lib.store.rdf.RDF_Store'>, 'sparql': <class 'kif_lib.store.sparql.SPARQL_Store'>, 'sparql-mapper': <class 'kif_lib.store.mapper.SPARQL_MapperStore'>, 'wikidata': <class 'kif_lib.store.wikidata.WikidataStore'>}

The store plugin registry.

store_name: ClassVar[str]

The name of this store plugin.

store_description: ClassVar[str]

The description of this store plugin.

exception Error

Bases: Error

Base class for store errors.

__init__(*args: Any, extra_references: TReferenceRecordSet | None = None, flags: Flags | None = None, limit: int | None = None, page_size: int | None = None, timeout: int | None = None, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

property context: Context

The current KIF context.

get_context(context: Context | None = None) → Context

Gets the current KIF context.

If context is not None, returns context.

Returns:

Context.

property default_extra_references: ReferenceRecordSet

The default value for Store.extra_references.

get_default_extra_references() → ReferenceRecordSet

Gets the default value for Store.extra_references.

Returns:

Reference record set.

property extra_references: ReferenceRecordSet

The extra references to attach to statements.

get_extra_references(default: ReferenceRecordSet | None = None) → ReferenceRecordSet

Gets the extra references to attach to statements.

If the extra references is None, returns default.

If default is None, assumes Store.default_extra_references.

Parameters:

default – Reference record set.

Returns:

Reference record set.

set_extra_references(extra_references: TReferenceRecordSet | None = None) → None

Sets the extra references to attach to statements.

If extra_references is None, assumes Store.default_extra_references.

Parameters:

references – Reference record set.

class Flags(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Flags

Store flags.

CACHE = 1

Whether to enable cache.

DEBUG = 2

Whether to enable debugging.

DISTINCT = 4

Whether to remove duplicates.

ORDER = 8

Whether to force some ordering.

BEST_RANK = 16

Whether to fetch only the best ranked statements.

VALUE_SNAK = 32

Whether to fetch value snaks.

SOME_VALUE_SNAK = 64

Whether to fetch some-value snaks.

NO_VALUE_SNAK = 128

Whether to fetch no-value snaks.

EARLY_FILTER = 256

Whether to enable early filtering.

LATE_FILTER = 512

Whether to enable late filtering.

ALL = 1023

All flags.

CACHE: Final[Flags] = 1

Whether to enable cache.

DEBUG: Final[Flags] = 2

Whether to enable debugging.

DISTINCT: Final[Flags] = 4

Whether to remove duplicates.

ORDER: Final[Flags] = 8

Whether to force some ordering.

BEST_RANK: Final[Flags] = 16

Whether to fetch only the best ranked statements.

VALUE_SNAK: Final[Flags] = 32

Whether to fetch value snaks.

SOME_VALUE_SNAK: Final[Flags] = 64

Whether to fetch some-value snaks.

NO_VALUE_SNAK: Final[Flags] = 128

Whether to fetch no-value snaks.

EARLY_FILTER: Final[Flags] = 256

Whether to enable early filtering.

LATE_FILTER: Final[Flags] = 512

Whether to enable late filtering.

property default_flags: Flags

The default value for Store.flags.

get_default_flags() → Flags

Gets the default value for Store.flags.

Returns:

Default store flags.

property flags: Flags

The store flags.

get_flags() → Flags

Gets the store flags.

Returns:

Store flags.

has_flags(flags: Flags) → bool

Tests whether flags are set in store.

Parameters:

flags – Store flags.

Returns:

True if successful; False otherwise.

set_flags(flags: Flags) → None

Sets flags in store.

Parameters:

flags – Store flags.

unset_flags(flags: Flags) → None

Unsets flags in store.

Parameters:

flags – Store flags.

property max_limit: int

The maximum value for Store.limit.

get_max_limit() → int

Gets the maximum value for Store.limit.

Returns:

Maximum limit.

property default_limit: int | None

The default value for Store.limit.

get_default_limit() → int | None

Gets the default value for Store.limit.

Returns:

Default limit or None.

property limit: int | None

The limit (maximum number) of responses.

get_limit(default: int | None = None) → int | None

Gets the limit (maximum number) of responses.

If the limit is None, returns default.

If default is None, assumes Store.default_limit.

Parameters:

default – Default limit.

Returns:

Limit or None.

set_limit(limit: int | None = None) → None

Sets the limit (maximum number) of responses.

If limit is negative, assumes zero.

If limit is None, assumes: attr: Store.default_limit.

Parameters:

limit – Limit.

property max_page_size: int

The maximum value for Store.page_size.

get_max_page_size() → int

Gets the maximum value for Store.page_size.

Returns:

Maximum page size.

property default_page_size: int

The default value for Store.page_size.

get_default_page_size() → int

Gets the default value for Store.page_size.

Returns:

Default page size.

property page_size: int

The page size of paginated responses.

get_page_size(default: int | None = None) → int

Gets the page size of paginated responses.

If the page size is None, returns default.

If default is None, assumes Store.default_page_size.

Parameters:

default – Page size.

Returns:

Page size.

set_page_size(page_size: int | None = None) → None

Sets page size of paginated responses.

If page_size is negative, assumes zero.

If page_size is None, assumes: attr: Store.default_page_size.

Parameters:

page_size – Page size.

property max_timeout: float

The maximum value for Store.timeout.

get_max_timeout() → float

Gets the maximum value for Store.timeout.

Returns:

Maximum timeout (in seconds).

property default_timeout: float | None

The default value for Store.timeout.

get_default_timeout() → float | None

Gets the default value for Store.timeout.

Returns:

Timeout or None.

property timeout: float | None

The timeout of responses (in seconds).

get_timeout(default: float | None = None) → float | None

Gets the timeout of responses (in seconds).

If the timeout is None, returns default.

If default is None, assumes Store.default_timeout.

Parameters:

default – Default timeout.

Returns:

Timeout or None.

set_timeout(timeout: float | None = None) → None

Sets the timeout of responses (in seconds).

If timeout is negative, assumes zero.

If timeout is None, assumes Store.default_timeout.

Parameters:

timeout – Timeout.

match(pattern: TPattern, limit: int | None = None, distinct: bool | None = None) → Iterator[ClosedTerm]

Searches for terms matching pattern.

Parameters:

• pattern – Pattern.

• limit – Limit (maximum number) of matches to return.

• distinct – Whether to skip duplicated matches.

Returns:

An iterator of closed-terms matching pattern.

contains(stmt: Statement) → bool

Tests whether statement is in store.

Parameters:

stmt – Statement.

Returns:

True if successful; False otherwise.

count(subject: TFingerprint | None = None, property: TFingerprint | None = None, value: TFingerprint | None = None, snak_mask: Filter.TSnakMask | None = None, snak: Snak | None = None, filter: Filter | None = None) → int

Counts statements matching filter.

Parameters:

• subject – Entity.

• property – Property.

• value – Value.

• snak_mask – Snak mask.

• snak – Snak.

• filter – Filter.

Returns:

The number of statements matching filter.

filter(subject: TFingerprint | None = None, property: TFingerprint | None = None, value: TFingerprint | None = None, snak_mask: Filter.TSnakMask | None = None, snak: Snak | None = None, filter: Filter | None = None, limit: int | None = None, distinct: bool | None = None) → Iterator[Statement]

Searches for statements matching filter.

Parameters:

• subject – Entity.

• property – Property.

• value – Value.

• snak_mask – Snak mask.

• snak – Snak.

• filter – Filter filter.

• limit – Limit (maximum number) of statements to return.

• distinct – Whether to skip duplicated matches.

Returns:

An iterator of statements matching filter.

filter_annotated(subject: TFingerprint | None = None, property: TFingerprint | None = None, value: TFingerprint | None = None, snak_mask: Filter.TSnakMask | None = None, snak: Snak | None = None, filter: Filter | None = None, limit: int | None = None) → Iterator[tuple[Statement, AnnotationRecordSet]]

Store.filter() with annotations.

Same as Store.filter() followed by Store.get_annotations().

Parameters:

• subject – Entity.

• property – Property.

• value – Value.

• snak_mask – Snak mask.

• snak – Snak.

• filter – Filter.

• limit – Maximum number of statements to return .

Returns:

An iterator of pairs “(statement, annotation record set)”.

get_annotations(stmts: Statement | Iterable[Statement]) → Iterator[tuple[Statement, AnnotationRecordSet | None]]

Gets annotation records of statements.

Parameters:

stmts – Statements.

Returns:

An iterator of pairs “(statement, annotation record set)”.

get_descriptor(entities: Entity | Iterable[Entity], language: str | None = None, mask: AttributeMask | int | None = None) → Iterator[tuple[Entity, Descriptor | None]]

Gets the descriptor of entities.

Parameters:

• entities – Entities.

• language – Language tag.

• mask – Descriptor mask.

Returns:

An iterator of pairs “(entity, descriptor)”.

get_item_descriptor(items: Item | Iterable[Item], language: str | None = None, mask: AttributeMask | int | None = None) → Iterator[tuple[Item, ItemDescriptor | None]]

Gets the descriptor of items.

Parameters:

• items – Items.

• language – Language tag.

• mask – Descriptor mask.

Returns:

An iterator of pairs “(item, descriptor)”.

get_property_descriptor(properties: Property | Iterable[Property], language: str | None = None, mask: AttributeMask | int | None = None) → Iterator[tuple[Property, PropertyDescriptor | None]]

Gets the descriptor of properties.

Parameters:

• properties – Properties.

• language – Language tag.

• mask – Descriptor mask.

Returns:

An iterator of pairs “(property, descriptor)”.

get_lexeme_descriptor(lexemes: Lexeme | Iterable[Lexeme], mask: AttributeMask | int | None = None) → Iterator[tuple[Lexeme, LexemeDescriptor | None]]

Gets the descriptor of lexemes.

Parameters:

lexemes – Lexemes.

Returns:

An iterator of pairs “(lexeme, descriptor)”.

class EmptyStore(store_name: str, *args: Any, **kwargs: Any)

Bases: Store

Empty store.

Parameters:

store_name – Name of the store plugin to instantiate.

__init__(store_name: str, *args, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

store_description: ClassVar[str] = 'Empty store'

The description of this store plugin.

store_name: ClassVar[str] = 'empty'

The name of this store plugin.

class SPARQL_Store(store_name: str, *args: Any, **kwargs: Any)

Bases: Store

SPARQL store.

Parameters:

• store_name – Store plugin to instantiate.

• iri – SPARQL endpoint IRI.

__init__(store_name: str, iri: IRI | String | str, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

set_timeout(timeout: float | None = None) → None

Sets the timeout of responses (in seconds).

If timeout is negative, assumes zero.

If timeout is None, assumes Store.default_timeout.

Parameters:

timeout – Timeout.

property iri: IRI

The IRI of the target SPARQL endpoint.

get_iri() → IRI

Gets the IRI of the target SPARQL endpoint.

Returns:

IRI.

store_description: ClassVar[str] = 'SPARQL endpoint'

The description of this store plugin.

store_name: ClassVar[str] = 'sparql'

The name of this store plugin.

class WikidataStore(store_name: str, *args: Any, **kwargs: Any)

Bases: SPARQL_Store

Wikidata store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• iri – Wikidata SPARQL endpoint IRI.

__init__(store_name: str, iri: T_IRI | None = None, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

store_description: ClassVar[str] = 'Wikidata store'

The description of this store plugin.

store_name: ClassVar[str] = 'wikidata'

The name of this store plugin.

class RDF_Store(store_name: str, *args: Any, **kwargs: Any)

Bases: SPARQL_Store

RDF store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Input sources, files, paths, or strings.

• publicID – Logical URI to use as the document base.

• format – Input source format (file extension or media type).

• location – Relative or absolute URL of the input source.

• file – File-like object to be used as input source.

• data – Data to be used as input source.

• graph – RDFLib graph to use.

• skolemize – Whether to skolemize the resulting graph.

__init__(store_name: str, *args: IO[bytes] | TextIO | InputSource | str | bytes | PurePath | None, publicID: str | None = None, format: str | None = None, location: str | None = None, file: BinaryIO | TextIO | None = None, data: str | bytes | None = None, graph: Graph | None = None, skolemize: bool = True, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

store_description: ClassVar[str] = 'RDF file'

The description of this store plugin.

store_name: ClassVar[str] = 'rdf'

The name of this store plugin.

class SPARQL_MapperStore(store_name: str, *args: Any, **kwargs: Any)

Bases: SPARQL_Store

SPARQL mapper store.

Parameters:

• store_name – Store plugin to instantiate.

• iri – SPARQL endpoint IRI.

• mapping – SPARQL mapping.

__init__(store_name: str, iri: T_IRI, mapping: SPARQL_Mapping, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

property mapping: SPARQL_Mapping

SPARQL mapping.

get_mapping() → SPARQL_Mapping

Gets SPARQL mapping.

Returns:

SPARQL mapping.

store_description: ClassVar[str] = 'SPARQL endpoint mapper'

The description of this store plugin.

store_name: ClassVar[str] = 'sparql-mapper'

The name of this store plugin.

class MixerStore(store_name: str, *args: Any, **kwargs: Any)

Bases: Store

Mixer store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• sources – Sources to mix.

• sync_flags – Whether to sync store flags.

__init__(store_name: str, sources: Iterable[Store] = (), sync_flags: bool = True, **kwargs: Any) → None

Initializes Store.

Parameters:

• store_name – Name of the store plugin to instantiate.

• args – Arguments to store plugin.

• extra_references – Extra references to attach to statements.

• flags – Store flags.

• limit – Limit (maximum number) of responses.

• page_size – Page size of paginated responses.

• timeout – Timeout of responses (in seconds).

• kwargs – Keyword arguments to store plugin.

property sources: Collection[Store]

The mixed sources.

get_sources() → Collection[Store]

Gets the mixed underlying sources.

Returns:

Mixed sources.

property sync_flags: bool

Whether to sync store flags.

get_sync_flags() → bool

Tests whether to sync store flags.

Returns:

True if successful; False otherwise.

store_description: ClassVar[str] = 'Mixer store'

The description of this store plugin.

store_name: ClassVar[str] = 'mixer'

The name of this store plugin.