Clients

`AsyncSolrClient`

Bases: BaseSolrClient[AsyncClient]

Asynchronous Python client for Apache Solr.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL of the Solr instance (e.g., "http://localhost:8983/solr")	required
`auth`	`Optional[SolrAuth]`	Authentication method to use (optional)	`None`
`timeout`	`float`	Request timeout in seconds	`10.0`
`verify`	`Union[bool, str]`	SSL certificate verification (default: True)	`True`
`**client_options`	`Any`	Additional options to pass to the httpx client	`{}`

Usage

from taiyo import AsyncSolrClient, BasicAuth
async with AsyncSolrClient(
    "http://localhost:8983/solr",
    auth=BasicAuth("username", "password")
) as client:
    client.set_collection("my_collection")
    results = await client.search("*:*")

Note: You must call set_collection() before using methods that require a collection.

`init(base_url, auth=None, timeout=10.0, verify=True, **client_options)`

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL of the Solr instance i.e. http://localhost:8983/solr.	required
`auth`	`Optional[SolrAuth]`	Authentication method to use (optional).	`None`
`timeout`	`float`	Request timeout in seconds. Defaults to 10.	`10.0`
`verify`	`Union[bool, str]`	SSL certificate verification. Can be True (default), False, or path to CA bundle.	`True`
`**client_options`	`Any`	Additional options to pass to the httpx client.	`{}`

`add(documents, commit=True)` `async`

Add one or more documents to the index.

Parameters:

Name	Type	Description	Default
`documents`	`Union[SolrDocument, List[SolrDocument]]`	A single document or list of documents to add. Can be dicts or instances of the document_model (which must be a subclass of SolrDocument).	required
`commit`	`bool`	Whether to commit the changes immediately	`True`

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr

`add_dynamic_field(field)` `async`

Add a dynamic field to the collection schema.

Parameters:

Name	Type	Description	Default
`field`	`Union[SolrDynamicField, Dict[str, Any]]`	SolrDynamicField instance or dictionary defining the dynamic field	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr Schema API

Example

from taiyo.schema import SolrDynamicField

field = SolrDynamicField(
    name="*_txt",
    type="text_general",
    indexed=True,
    stored=True
)
await client.add_dynamic_field(field)

`add_field(field)` `async`

Add a field to the collection schema.

Parameters:

Name	Type	Description	Default
`field`	`Union[SolrField, Dict[str, Any]]`	SolrField instance or dictionary defining the field	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr Schema API

Example

from taiyo.schema import SolrField

field = SolrField(
    name="title",
    type="text_general",
    indexed=True,
    stored=True
)
await client.add_field(field)

`add_field_type(field_type)` `async`

Add a field type to the collection schema.

Parameters:

Name	Type	Description	Default
`field_type`	`Union[SolrFieldType, Dict[str, Any]]`	SolrFieldType instance or dictionary defining the field type	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr Schema API

Example

from taiyo.schema import SolrFieldType, SolrFieldClass

field_type = SolrFieldType(
    name="text_general",
    solr_class=SolrFieldClass.TEXT,
    position_increment_gap=100
)
await client.add_field_type(field_type)

`close()` `async`

Close the underlying HTTP client.

`commit()` `async`

Commit pending changes to the index.

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr

`create_collection(name, num_shards=1, replication_factor=1, **kwargs)` `async`

Create a new Solr collection using the v2 Collections API (POST /api/collections).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the collection to create.	required
`num_shards`	`int`	Number of shards for the collection.	`1`
`replication_factor`	`int`	Replication factor for the collection.	`1`
`**kwargs`	`Any`	Additional Solr parameters (included in JSON body).	`{}`

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr.

`delete(query=None, ids=None, commit=True)` `async`

Delete documents from the index.

Supports multiple delete formats per Apache Solr specification: - Single ID: {"delete": "myid"} - Multiple IDs: {"delete": ["id1", "id2"]} - Query: {"delete": {"query": "field:value"}} - Combined: {"delete": {"id": "myid", "query": "field:value"}}

Parameters:

Name	Type	Description	Default
`query`	`Optional[str]`	Delete documents matching this query	`None`
`ids`	`Optional[Union[str, List[str]]]`	Delete document(s) with this ID or these IDs. Can be a single string or list.	`None`
`commit`	`bool`	Whether to commit the changes immediately	`True`

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr

Example

# Delete by single ID
await client.delete(ids="doc1")

# Delete by multiple IDs
await client.delete(ids=["doc1", "doc2"])

# Delete by query
await client.delete(query="status:archived")

# Delete by ID and query combined
await client.delete(ids="doc1", query="status:archived")

`delete_collection(name, **kwargs)` `async`

Delete a Solr collection.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the collection to delete.	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr.

`ping()` `async`

Ping the Solr instance to check if it's available.

Returns:

Type	Description
`bool`	True if Solr is available, False otherwise

`search(query, document_model=SolrDocument, **kwargs)` `async`

Search the Solr index.

Parameters:

Name	Type	Description	Default
`query`	`Union[str, Dict[str, Any], BaseQueryParser]`	Query string, dictionary of parameters, or QueryBuilder instance	required
`document_model`	`Type[DocumentT]`	Pydantic model class for documents	`SolrDocument`
`**kwargs`	`Any`	Additional query parameters	`{}`

Returns:

Type	Description
`SolrResponse[DocumentT]`	SolrResponse with docs as a list of document_model instances (subclass of SolrDocument).

`BasicAuth`

Bases: SolrAuth

Basic HTTP authentication. https://solr.apache.org/guide/solr/latest/deployment-guide/basic-authentication-plugin.html

Parameters:

Name	Type	Description	Default
`username`	`str \| SecretStr`	Username for authentication (str or SecretStr)	required
`password`	`str \| SecretStr`	Password for authentication (str or SecretStr)	required

Example

auth = BasicAuth("admin", "secret")
client = SolrClient("http://localhost:8983/solr", "my_collection", auth=auth)

`BearerAuth`

Bases: SolrAuth

Bearer token authentication. https://solr.apache.org/guide/solr/latest/deployment-guide/jwt-authentication-plugin.html

Parameters:

Name	Type	Description	Default
`token`	`str \| SecretStr`	JWT token to use (str or SecretStr)	required

Example

auth = BearerAuth("my-token")
client = SolrClient("http://localhost:8983/solr", "my_collection", auth=auth)

`OAuth2Auth`

Bases: SolrAuth

OAuth 2.0 authentication with token refresh.

Parameters:

Name	Type	Description	Default
`client_id`	`str \| SecretStr`	OAuth client ID (str or SecretStr)	required
`client_secret`	`str \| SecretStr`	OAuth client secret (str or SecretStr)	required
`token_url`	`str`	Token endpoint URL	required

Example

auth = OAuth2Auth(
    client_id="your-client-id",
    client_secret="your-client-secret",
    token_url="https://auth.example.com/token"
)
client = SolrClient("http://localhost:8983/solr", auth=auth)

`get_access_token()`

Fetch access token from OAuth2 server.

Returns:

Type	Description
`str`	Access token string

Raises:

Type	Description
`HTTPStatusError`	If token request fails

`SolrAuth`

Base class for Solr authentication methods.

`apply(client)`

Apply authentication to the client.

Parameters:

Name	Type	Description	Default
`client`	`BaseSolrClient[Any]`	The httpx.AsyncClient instance to apply authentication to	required

`SolrClient`

Bases: BaseSolrClient[Client]

Synchronous Python client for Apache Solr.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL of the Solr instance (e.g., "http://localhost:8983/solr")	required
`auth`	`Optional[SolrAuth]`	Authentication method to use (optional)	`None`
`timeout`	`float`	Request timeout in seconds	`10.0`
`verify`	`Union[bool, str]`	SSL certificate verification (default: True)	`True`
`**client_options`	`Any`	Additional options to pass to the httpx client	`{}`

Usage

from taiyo import SolrClient, BasicAuth
with SolrClient(
    "http://localhost:8983/solr",
    auth=BasicAuth("username", "password")
) as client:
    client.set_collection("my_collection")
    results = client.search("*:*")

Note: You must call set_collection() before using methods that require a collection.

`init(base_url, auth=None, timeout=10.0, verify=True, **client_options)`

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL of the Solr instance i.e. http://localhost:8983/solr.	required
`auth`	`Optional[SolrAuth]`	Authentication method to use (optional).	`None`
`timeout`	`float`	Request timeout in seconds. Defaults to 10.	`10.0`
`verify`	`Union[bool, str]`	SSL certificate verification. Can be True (default), False, or path to CA bundle.	`True`
`**client_options`	`Any`	Additional options to pass to the httpx client.	`{}`

`add(documents, commit=True)`

Add one or more documents to the index.

Parameters:

Name	Type	Description	Default
`documents`	`Union[SolrDocument, List[SolrDocument]]`	A single document or list of documents to add. Can be dicts or instances of the document_model (which must be a subclass of SolrDocument).	required
`commit`	`bool`	Whether to commit the changes immediately	`True`

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr

`add_dynamic_field(field)`

Add a dynamic field to the collection schema.

Parameters:

Name	Type	Description	Default
`field`	`Union[SolrDynamicField, Dict[str, Any]]`	SolrDynamicField instance or dictionary defining the dynamic field	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr Schema API

Example

from taiyo.schema import SolrDynamicField

field = SolrDynamicField(
    name="*_txt",
    type="text_general",
    indexed=True,
    stored=True
)
client.add_dynamic_field(field)

`add_field(field)`

Add a field to the collection schema.

Parameters:

Name	Type	Description	Default
`field`	`Union[SolrField, Dict[str, Any]]`	SolrField instance or dictionary defining the field	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr Schema API

Example

from taiyo.schema import SolrField

field = SolrField(
    name="title",
    type="text_general",
    indexed=True,
    stored=True
)
client.add_field(field)

`add_field_type(field_type)`

Add a field type to the collection schema.

Parameters:

Name	Type	Description	Default
`field_type`	`Union[SolrFieldType, Dict[str, Any]]`	SolrFieldType instance or dictionary defining the field type	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr Schema API

Example

from taiyo.schema import SolrFieldType, SolrFieldClass

field_type = SolrFieldType(
    name="text_general",
    solr_class=SolrFieldClass.TEXT,
    position_increment_gap=100
)
client.add_field_type(field_type)

`close()`

Close the underlying HTTP client.

`commit()`

Commit pending changes to the index.

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr

`create_collection(name, num_shards=1, replication_factor=1, **kwargs)`

Create a new Solr collection using the v2 Collections API (POST /api/collections).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the collection to create.	required
`num_shards`	`int`	Number of shards for the collection.	`1`
`replication_factor`	`int`	Replication factor for the collection.	`1`
`**kwargs`	`Any`	Additional Solr parameters (included in JSON body).	`{}`

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr.

`delete(query=None, ids=None, commit=True)`

Delete documents from the index.

Supports multiple delete formats per Apache Solr specification: - Single ID: {"delete": "myid"} - Multiple IDs: {"delete": ["id1", "id2"]} - Query: {"delete": {"query": "field:value"}} - Combined: {"delete": {"id": "myid", "query": "field:value"}}

Parameters:

Name	Type	Description	Default
`query`	`Optional[str]`	Delete documents matching this query	`None`
`ids`	`Optional[Union[str, List[str]]]`	Delete document(s) with this ID or these IDs. Can be a single string or list.	`None`
`commit`	`bool`	Whether to commit the changes immediately	`True`

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr

Example

# Delete by single ID
client.delete(ids="doc1")

# Delete by multiple IDs
client.delete(ids=["doc1", "doc2"])

# Delete by query
client.delete(query="status:archived")

# Delete by ID and query combined
client.delete(ids="doc1", query="status:archived")

`delete_collection(name)`

Delete a Solr collection.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the collection to delete.	required

Returns:

Type	Description
`Dict[str, Any]`	Response from Solr.

`ping()`

Ping the Solr instance to check if it's available.

Returns:

Type	Description
`bool`	True if Solr is available, False otherwise

`search(query, document_model=SolrDocument, **kwargs)`

Search the Solr index.

Parameters:

Name	Type	Description	Default
`query`	`Union[str, Dict[str, Any], BaseQueryParser]`	query string, dictionary of parameters, or BaseQueryParser instance	required
`document_model`	`Type[DocumentT]`	Pydantic model class for documents	`SolrDocument`
`**kwargs`	`Any`	Additional query parameters	`{}`

Returns:

Type	Description
`SolrResponse[DocumentT]`	SolrResponse with docs as a list of document_model instances (subclass of SolrDocument).

Clients

AsyncSolrClient

__init__(base_url, auth=None, timeout=10.0, verify=True, **client_options)

add(documents, commit=True) async

add_dynamic_field(field) async

add_field(field) async

add_field_type(field_type) async

close() async

commit() async

create_collection(name, num_shards=1, replication_factor=1, **kwargs) async

delete(query=None, ids=None, commit=True) async

delete_collection(name, **kwargs) async

ping() async

search(query, document_model=SolrDocument, **kwargs) async

BasicAuth

BearerAuth

OAuth2Auth

get_access_token()

SolrAuth

apply(client)

SolrClient

__init__(base_url, auth=None, timeout=10.0, verify=True, **client_options)

add(documents, commit=True)

add_dynamic_field(field)

add_field(field)

add_field_type(field_type)

close()

commit()

create_collection(name, num_shards=1, replication_factor=1, **kwargs)

delete(query=None, ids=None, commit=True)

delete_collection(name)

ping()

search(query, document_model=SolrDocument, **kwargs)

`AsyncSolrClient`

`init(base_url, auth=None, timeout=10.0, verify=True, **client_options)`

`add(documents, commit=True)` `async`

`add_dynamic_field(field)` `async`

`add_field(field)` `async`

`add_field_type(field_type)` `async`

`close()` `async`

`commit()` `async`

`create_collection(name, num_shards=1, replication_factor=1, **kwargs)` `async`

`delete(query=None, ids=None, commit=True)` `async`

`delete_collection(name, **kwargs)` `async`

`ping()` `async`

`search(query, document_model=SolrDocument, **kwargs)` `async`

`BasicAuth`

`BearerAuth`

`OAuth2Auth`

`get_access_token()`

`SolrAuth`

`apply(client)`

`SolrClient`

`init(base_url, auth=None, timeout=10.0, verify=True, **client_options)`

`add(documents, commit=True)`

`add_dynamic_field(field)`

`add_field(field)`

`add_field_type(field_type)`

`close()`

`commit()`

`create_collection(name, num_shards=1, replication_factor=1, **kwargs)`

`delete(query=None, ids=None, commit=True)`

`delete_collection(name)`

`ping()`

`search(query, document_model=SolrDocument, **kwargs)`