Query Utility for Elasticsearch
Project description
Query Utility for Elasticsearch
Utility library for creating elasticsearch query proxies using FastAPI.
from fastapi_elasticsearch import ElasticsearchAPIRouteBuilder
route_builder = ElasticsearchAPIRouteBuilder(
# The elasticsearch client
es_client=elasticsearch_client,
# The index or indices that your query will run.
index_name=index_name
)
# Decorate a function as a filter.
# The filter can declare parameters.
@route_builder.filter()
def filter_category(c: Optional[str] = Query(None)):
return {
"term": {
"category": c
}
} if c is not None else None
# Decorate a function as a matcher
# (will contribute to the query scoring).
# Parameters can also be used.
@route_builder.matcher()
def match_fields(q: Optional[str] = Query(None)):
return {
"multi_match": {
"query": q,
"fuzziness": "AUTO",
"fields": [
"name^2",
"description"
]
}
} if q is not None else None
# Decorate a function as a sorter.
# Parameters can be declared.
@route_builder.sorter()
def sort_by(direction: Optional[str] = Query(None)):
return {
"name": direction
} if direction is not None else None
# Decorate a function as a highlighter.
# Parameters can also be declared.
@route_builder.highlighter()
def highlight(q: Optional[str] = Query(None),
h: bool = Query(False):
return {
"name": {}
} if q is not None and h else None
app = FastAPI()
# Add the route to the app using the default endpoint.
es_route = route_builder.build("/search")
app.routes.append(es_route)
# It is possible to customize the route endpoint.
@route_builder.endpoint("/search")
async def search(query_body = Depends(route_builder.query_builder)) -> JSONResponse:
response = es_client.search(
body=query_body,
index=index_name
)
modified_response = modify_response(response)
return modified_response
# And use different parameters
@route_builder.endpoint("/search-more")
async def get_document(
req: Request,
size: Optional[int] = Query(100,
le=1000,
alias="s",
description="Defines the number of hits to return."),
start_from: Optional[int] = Query(0,
alias="f",
description="Starting document offset.")) -> JSONResponse:
query_body = route_builder.query_builder.build_body(
request=req,
size=size,
start_from=start_from,
)
return es_client.search(
body=query_body,
index=index_name
)
It is possible to customize the generated query body using the decorator @search_builder.
from typing import List, Dict
@es_router.search_builder()
def build_search_body(size: int = 10,
start_from: int = 0,
scroll: str = None,
filters: List[Dict] = [],
matchers: List[Dict] = [],
highlighters: List[Dict] = [],
sorters: List[Dict] = []) -> Dict:
return {
"query": {
...
},
...
}
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for fastapi-elasticsearch-0.4.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | b18bde1d14051e6d1932c8e555aca6a239d9c68ce3f79d441c03c673c6350f4d |
|
MD5 | 091055d8a361bd42c664a8dd2d98c48b |
|
BLAKE2b-256 | 0c7c236086b8e5c8dc2f060d1b4ac40a16c3f8ccff03eca1c4494526ef6baad4 |
Close
Hashes for fastapi_elasticsearch-0.4.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | ce42f2b7f820b26b9bc8d86d89ad97f97ab2f0aaeea6c66e48e9749cf918cf6a |
|
MD5 | 505a198aa900b49c4bf72a1ab4d40b25 |
|
BLAKE2b-256 | 5505cac3cc2c57e1ae559154d9197baaf87610c22e13260aa8815a82199e680a |