Notes and best practices

There are multiple ways to connect to your Weaviate instance. To instantiate a client, use one of these styles:

• Connection helper functions
• Explicit instantiation

• weaviate.connect_to_weaviate_cloud()
  • Previously connect_to_wcs()
• weaviate.connect_to_local()
• weaviate.connect_to_embedded()
• weaviate.connect_to_custom()

import weaviate
from weaviate.classes.init import Auth
import os

# Best practice: store your credentials in environment variables
weaviate_url = os.environ["WEAVIATE_URL"]
weaviate_api_key = os.environ["WEAVIATE_API_KEY"]
openai_api_key = os.environ["OPENAI_API_KEY"]

client = weaviate.connect_to_weaviate_cloud(
    cluster_url=weaviate_url,  # Replace with your Weaviate Cloud URL
    auth_credentials=Auth.api_key(weaviate_api_key),  # Replace with your Weaviate Cloud key
    headers={'X-OpenAI-Api-key': openai_api_key}  # Replace with your OpenAI API key
)

import weaviate

client = weaviate.connect_to_local()  # Connect with default parameters

import weaviate

client = weaviate.connect_to_embedded()  # Connect with default parameters

import weaviate

client = weaviate.connect_to_custom(
    http_host="localhost",
    http_port=8080,
    http_secure=False,
    grpc_host="localhost",
    grpc_port=50051,
    grpc_secure=False,
    headers={
        "X-OpenAI-Api-Key": os.getenv("OPENAI_API_KEY")  # Or any other inference API keys
    }
)

The v4 client helper functions provide some optional parameters to customize your client.

• Specify external API keys
• Specify connection timeout values
• Specify authentication details

To add API keys for services such as Cohere or OpenAI, use the headers parameter.

import weaviate
import os

client = weaviate.connect_to_local(
    headers={
        "X-OpenAI-Api-Key": os.getenv("OPENAI_API_KEY")
    }
)

You can set timeout values, in seconds, for the client. Use the Timeout class to configure the timeout values for initialization checks as well as query and insert operations.

import weaviate
from weaviate.classes.init import AdditionalConfig, Timeout

client = weaviate.connect_to_local(
    port=8080,
    grpc_port=50051,
    additional_config=AdditionalConfig(
        timeout=Timeout(init=30, query=60, insert=120)  # Values in seconds
    )
)

Some of the connect helper functions take authentication credentials. For example, connect_to_weaviate_cloud accepts a WCD API key or OIDC authentication credentials.

import weaviate
from weaviate.classes.init import Auth
import os

# Best practice: store your credentials in environment variables
weaviate_url = os.environ["WEAVIATE_URL"]
weaviate_api_key = os.environ["WEAVIATE_API_KEY"]
openai_api_key = os.environ["OPENAI_API_KEY"]

client = weaviate.connect_to_weaviate_cloud(
    cluster_url=weaviate_url,  # Replace with your Weaviate Cloud URL
    auth_credentials=Auth.api_key(weaviate_api_key),  # Replace with your Weaviate Cloud key
    headers={'X-OpenAI-Api-key': openai_api_key}  # Replace with your OpenAI API key
)

import weaviate

client = weaviate.connect_to_weaviate_cloud(
    cluster_url=os.getenv("WEAVIATE_URL"),  # Replace with your Weaviate Cloud URL
    auth_credentials=weaviate.auth.AuthClientPassword(
        username=os.getenv("WCD_USERNAME"),  # Your Weaviate Cloud username
        password=os.getenv("WCD_PASSWORD")   # Your Weaviate Cloud password
    )
)

For OIDC authentication with the Client Credentials flow, use the AuthClientCredentials class.

For OIDC authentication with the Refresh Token flow, use the AuthBearerToken class.

If the helper functions do not provide the customization you need, use the WeaviateClient class to instantiate the client.

If you need to pass custom parameters, use the weaviate.WeaviateClient class to instantiate a client. This is the most flexible way to instantiate the client object.

When you instantiate a connection directly, you have to call the .connect() method to connect to the server.

import weaviate
from weaviate.connect import ConnectionParams
from weaviate.classes.init import AdditionalConfig, Timeout, Auth
import os

client = weaviate.WeaviateClient(
    connection_params=ConnectionParams.from_params(
        http_host="localhost",
        http_port=8099,
        http_secure=False,
        grpc_host="localhost",
        grpc_port=50052,
        grpc_secure=False,
    ),
    auth_client_secret=Auth.api_key("secr3tk3y"),
    additional_headers={
        "X-OpenAI-Api-Key": os.getenv("OPENAI_API_KEY")
    },
    additional_config=AdditionalConfig(
        timeout=Timeout(init=30, query=60, insert=120),  # Values in seconds
    ),
    skip_init_checks=False
)

client.connect()  # When directly instantiating, you need to connect manually

The Python client doesn't directly support passing SSL certificates. If you need to work with self-signed certificates (e.g. for enterprise environments), you have two options:

You can add the custom SSL certificates to the underlying libraries such as certifi that the Weaviate client library uses.

Alternatively, you can set the environment variables GRPC_DEFAULT_SSL_ROOTS_FILE_PATH and SSL_CERT_FILE to the path of the certificate file. At instantiation, also set additional_config=AdditionalConfig(trust_env=True). Otherwise, the client library will not use the environment variables.

import os
import weaviate
from weaviate.classes.init import AdditionalConfig

# Set environment variables for SSL certificates
# Set it here or in your shell (e.g. .bashrc or .zshrc file)
os.environ["GRPC_DEFAULT_SSL_ROOTS_FILE_PATH"] = "/path/to/your/cert.crt"
os.environ["SSL_CERT_FILE"] = "/path/to/your/cert.crt"

# Then connect to Weaviate
client = weaviate.connect_to_custom(
    http_host=weaviate_host,  # Replace with your Weaviate host
    http_port=8080,
    http_secure=True,
    grpc_host=weaviate_grpc_host,  # Replace with your Weaviate gRPC host
    grpc_port=50051,
    grpc_secure=True,
    additional_config=AdditionalConfig(trust_env=True)  # Required for custom SSL certificates
)

When establishing a connection to the Weaviate server, the client performs a series of checks. These includes checks for the server version, and to make sure that the REST and gRPC ports are available.

You can set skip_init_checks to True to skip these checks.

import weaviate

client = weaviate.connect_to_local(
    skip_init_checks=True
)

In most cases, you should use the default False setting for skip_init_checks. However, setting skip_init_checks=True may be a useful temporary measure if you have connection issues.

For additional connection configuration, see Timeout values.

The idiomatic way to create a collection object is client.collections.use(<COLLECTION_NAME>). While identical to client.collections.get(), use() is more clearly indicative of the fact that it does not perform any network requests.

We made this change as client.collections.get() may be misinterpreted as fetching the collection schema from the server, which it does not.

In the future, client.collections.get() may be deprecated.

The v4 client offers two ways to perform batch imports. From the client object directly, or from the collection object.

We recommend using the collection object to perform batch imports of single collections or tenants. If you are importing objects across many collections, such as in a multi-tenancy configuration, using client.batch may be more convenient.

There are four methods to configure the batching behavior. They are stream, dynamic, fixed_size and rate_limit.

We recommend using a context manager as shown below.

These methods return a new context manager for each batch. Attributes that are returned from one batch, such as failed_objects or failed_references, are not included in any subsequent calls.

import weaviate

client = weaviate.connect_to_local()

try:
    with client.batch.dynamic() as batch:  # or <collection>.batch.dynamic()
        # Batch import objects/references - e.g.:
        batch.add_object(properties={"title": "Multitenancy"}, collection="WikiArticle", uuid=src_uuid)
        batch.add_object(properties={"title": "Database schema"}, collection="WikiArticle", uuid=tgt_uuid)
        batch.add_reference(from_collection="WikiArticle", from_uuid=src_uuid, from_property="linkedArticle", to=tgt_uuid)

finally:
    client.close()

import weaviate

client = weaviate.connect_to_local()

try:
    with client.batch.stream() as batch:  # or <collection>.batch.stream()
        # Batch import objects/references - e.g.:
        batch.add_object(properties={"title": "Multitenancy"}, collection="WikiArticle", uuid=src_uuid)
        batch.add_object(properties={"title": "Database schema"}, collection="WikiArticle", uuid=tgt_uuid)
        batch.add_reference(from_collection="WikiArticle", from_uuid=src_uuid, from_property="linkedArticle", to=tgt_uuid)

finally:
    client.close()

import weaviate

client = weaviate.connect_to_local()

try:
    with client.batch.fixed_size(batch_size=100, concurrent_requests=4) as batch:  # or <collection>.batch.fixed_size()
        # Batch import objects/references - e.g.:
        batch.add_object(properties={"title": "Multitenancy"}, collection="WikiArticle", uuid=src_uuid)
        batch.add_object(properties={"title": "Database schema"}, collection="WikiArticle", uuid=tgt_uuid)
        batch.add_reference(from_collection="WikiArticle", from_uuid=src_uuid, from_property="linkedArticle", to=tgt_uuid)

finally:
    client.close()

import weaviate

client = weaviate.connect_to_local()

try:
    with client.batch.rate_limit(requests_per_minute=600) as batch:  # or <collection>.batch.rate_limit()
        # Batch import objects/references - e.g.:
        batch.add_object(properties={"title": "Multitenancy"}, collection="WikiArticle", uuid=src_uuid)
        batch.add_object(properties={"title": "Database schema"}, collection="WikiArticle", uuid=tgt_uuid)
        batch.add_reference(from_collection="WikiArticle", from_uuid=src_uuid, from_property="linkedArticle", to=tgt_uuid)

finally:
    client.close()

If the background thread that is responsible for sending the batches raises an exception during batch processing, the error is raised to the main thread.

collection.data.ingest(objs) is a one-shot convenience that uses server-side batching under the hood (no batching context required). It accepts any iterable of plain property dicts or DataObject instances, and returns the same BatchObjectReturn object as insert_many. Pass a list of objects that you already hold in memory to use ingest as a drop-in replacement for insert_many on large lists. Pass a generator, or any other lazy iterable, to import from a source that does not fit in memory: the client sends each object to the server as the generator produces it.

import weaviate

client = weaviate.connect_to_local()

# A generator produces objects one at a time instead of building a list
def article_titles():
    for title in ["Multitenancy", "Database schema"]:
        yield {"title": title}

try:
    articles = client.collections.use("WikiArticle")
    # `ingest` accepts any iterable, including a generator
    result = articles.data.ingest(article_titles())

    if result.errors:
        print(f"Number of failed imports: {len(result.errors)}")

finally:
    client.close()

For a generator that reads a source file line by line, see Batch import.

During a batch import, any failed objects or references will be stored for retrieval. Additionally, a running count of failed objects and references is maintained.

The counter can be accessed through batch.number_errors within the context manager.

A list of failed objects can be obtained through batch.failed_objects and a list of failed references can be obtained through batch.failed_references.

Note that these lists are reset when a batching process is initialized. So make sure to retrieve them before starting a new batch import block.

import weaviateclient = weaviate.connect_to_local()try:    # ===== First batch import block =====    with client.batch.rate_limit(requests_per_minute=600) as batch:  # or <collection>.batch.rate_limit()        # Batch import objects/references        for i in source_iterable:  # Some insertion loop            if batch.number_errors > 10:  # Monitor errors during insertion                # Break or raise an exception                pass    # Note these are outside the `with` block - they are populated after the context manager exits    failed_objs_a = client.batch.failed_objects  # Get failed objects from the first batch import    failed_refs_a = client.batch.failed_references  # Get failed references from the first batch import    # ===== Second batch import block =====    # This will clear the failed objects/references    with client.batch.rate_limit(requests_per_minute=600) as batch:  # or <collection>.batch.rate_limit()        # Batch import objects/references        for i in source_iterable:  # Some insertion loop            if batch.number_errors > 10:  # Monitor errors during insertion                # Break or raise an exception                pass    # Note these are outside the `with` block - they are populated after the context manager exits    failed_objs_b = client.batch.failed_objects  # Get failed objects from the second batch import    failed_refs_b = client.batch.failed_references  # Get failed references from the second batch importfinally:    client.close()

collection.data.ingest() does not use a batching context, so it reports failures through its return value instead. Check result.has_errors for a quick summary flag that tells you whether anything failed. For the detail, check result.errors, a dictionary that holds one entry per failed object, keyed by the position of the object in the input. The one-shot ingest example above shows this pattern.

Some model providers provide batch vectorization APIs, where each request can include multiple objects.

From Weaviate v1.25.0, a batch import automatically makes use of the model providers' batch vectorization APIs where available. This reduces the number of requests to the model provider, improving throughput.

The client automatically handles vectorization if you set the vectorizer when you create the collection.

collection = client.collections.create(
        name="NewCollection",
        properties=[
            Property(name="url", data_type=DataType.TEXT),
            Property(name="title", data_type=DataType.TEXT),
            Property(name="raw", data_type=DataType.TEXT),
            Property(name="sha", data_type=DataType.TEXT),
        ],
        vector_config=[
            Configure.Vectors.text2vec_cohere(name="cohereFirst"),
            Configure.Vectors.text2vec_cohere(name="cohereSecond"),
        ]
    )

To modify the vectorization settings, update the client object. This example adds multiple vectorizers:

• Cohere. Set the service API key. Set the request rate.
• OpenAI. Set the service API key. Set the base URL.
• VoyageAI. Set the service API key.

from weaviate.classes.config import Integrations

integrations = [
    # Each model provider may expose different parameters
    Integrations.cohere(
        api_key=cohere_key,
        requests_per_minute_embeddings=rpm_embeddings,
    ),
    Integrations.openai(
        api_key=openai_key,
        requests_per_minute_embeddings=rpm_embeddings,
        tokens_per_minute_embeddings=tpm_embeddings,   # e.g. OpenAI also exposes tokens per minute for embeddings
    ),
]
client.integrations.configure(integrations)

The client library provides numerous additional Python classes to provide IDE assistance and typing help. You can import them individually, like so:

from weaviate.classes.config import Property, ConfigFactory
from weaviate.classes.data import DataObject
from weaviate.classes.query import Filter

But it may be convenient to import the whole set of classes like this. You will see both usage styles in our documentation.

import weaviate.classes as wvc

For discoverability, the classes are arranged into submodules.

You must ensure your client connections are closed. You can use client.close(), or use a context manager to close client connections for you.

This will close the client connection when the try block is complete (or if an exception is raised).

import weaviate

client = weaviate.connect_to_local()  # Connect with default parameters

try:
    pass  # Do something with the client

finally:
    client.close()  # Ensure the connection is closed

This will close the client connection when you leave the with block.

import weaviate

with weaviate.connect_to_local() as client:
    # Do something with the client
    pass
    # The connection is closed automatically when the context manager exits

The client library raises exceptions for various error conditions. These include, for example:

• weaviate.exceptions.WeaviateConnectionError for failed connections.
• weaviate.exceptions.WeaviateQueryError for failed queries.
• weaviate.exceptions.WeaviateBatchError for failed batch operations.
• weaviate.exceptions.WeaviateClosedClientError for operations on a closed client.

Each of these exceptions inherit from weaviate.exceptions.WeaviateBaseError, and can be caught using this base class, as shown below.

try:
    collection = client.collections.use("NonExistentCollection")
    collection.query.fetch_objects(limit=2)
except weaviate.exceptions.WeaviateBaseError as e:
    print(f"Caught a Weaviate error: {e.message}")

You can review this module which defines the exceptions that can be raised by the client library.

The client library doc strings also provide information on the exceptions that can be raised by each method. You can view these by using the help function in Python, by using the ? operator in Jupyter notebooks, or by using an IDE, such as hover-over tooltips in VSCode.

While the Python client is fundamentally designed to be thread-safe, it's important to note that due to its dependency on the requests library, complete thread safety isn't guaranteed.

This is an area that we are looking to improve in the future.

If you are performing batching in a multi-threaded scenario, ensure that only one of the threads is performing the batching workflow at any given time. No two threads can use the same client.batch object at one time.

Each query response object typically include multiple attributes. Consider this query.

questions = client.collections.use("JeopardyQuestion")
response = questions.generate.near_text(
    query="history",
    limit=2,
    single_prompt="Translate this into French {question}",
    grouped_task="Summarize this into a sentence",
    return_metadata=wvc.query.MetadataQuery(
        distance=True,
        creation_time=True
    )
)

print("Grouped Task generated outputs:")
print(response.generative.text)
for o in response.objects:
    print(f"Outputs for object {o.uuid}")
    print(f"Generated text:")
    print(o.generative.text)
    print(f"Properties:")
    print(o.properties)
    print(f"Metadata")
    print(o.metadata)

Each response includes attributes such as objects and generated. Then, each object in objects include multiple attributes such as uuid, vector, properties, references, metadata and generated.

_GenerativeReturn(objects=[_GenerativeObject(uuid=UUID('61e29275-8f53-5e28-a355-347d45a847b3'), metadata=_MetadataReturn(creation_time=datetime.datetime(2024, 1, 2, 18, 3, 7, 475000, tzinfo=datetime.timezone.utc), last_update_time=None, distance=0.19253945350646973, certainty=None, score=None, explain_score=None, is_consistent=None, rerank_score=None), properties={'points': 1000.0, 'answer': 'Daniel Boorstein', 'air_date': datetime.datetime(1990, 3, 26, 0, 0, tzinfo=datetime.timezone.utc), 'round': 'Double Jeopardy!', 'question': 'This historian & former Librarian of Congress was teaching history at Harvard while studying law at Yale'}, references=None, vector=None, generated="Cet historien et ancien bibliothécaire du Congrès enseignait l'histoire à Harvard tout en étudiant le droit à Yale."), _GenerativeObject(uuid=UUID('e987d1a1-2599-5dd8-bd22-4f3b0338539a'), metadata=_MetadataReturn(creation_time=datetime.datetime(2024, 1, 2, 18, 3, 8, 185000, tzinfo=datetime.timezone.utc), last_update_time=None, distance=0.193121075630188, certainty=None, score=None, explain_score=None, is_consistent=None, rerank_score=None), properties={'points': 400.0, 'air_date': datetime.datetime(2007, 5, 11, 0, 0, tzinfo=datetime.timezone.utc), 'answer': 'an opinion', 'round': 'Jeopardy!', 'question': 'This, a personal view or belief, comes from the Old French for "to think"'}, references=None, vector=None, generated='Ceci, une opinion personnelle ou une croyance, provient du vieux français signifiant "penser".')], generated='Daniel Boorstein, a historian and former Librarian of Congress, taught history at Harvard while studying law at Yale, and an opinion is a personal view or belief derived from the Old French word for "to think".')

To limit the response payload, you can specify which properties and metadata to return.

The client library performs input argument validation by default to make sure that the input types match the expected types.

You can disable this validation to improve performance. You can do this by setting the skip_argument_validation parameter to True when you instantiate a collection object, with collections.get, or with collections.create for example.

# Configure the `performant_articles` to skip argument validation on its methods
performant_articles = client.collections.use("Article", skip_argument_validation=True)

This may be useful in cases where you are using the client library in a production environment, where you can be confident that the input arguments are typed correctly.

If you use a browser to run the Python client with a Jupyter notebook, press Tab for code completion while you edit. If you use VSCode to run your Jupyter notebook, press control + space for code completion.

To provide raw GraphQL queries, you can use the client.graphql_raw_query method (previously client.query.raw in the v3 client). This method takes a string as input.