### Setup Index Mappings

Source: https://context7.com/wallneradam/esorm/llms.txt

Configures ElasticSearch index templates and mappings for your defined models.

```APIDOC
## Setup Index Mappings

### Description
Configures ElasticSearch index templates and mappings for your defined models.

### Method
`setup_mappings` function

### Endpoint
N/A (Setup function)

### Parameters
#### Arguments
- **models** (module or list[ESModel]) - Required - The module or list containing your ESORM models.
- **prefix_name** (str) - Optional - Sets a custom prefix for all index names. Overrides default or `set_default_index_prefix`.
- **shards** (int) - Optional - Number of primary shards for the index template.
- **auto_expand_replicas** (str) - Optional - Defines the range for auto-expanding replicas (e.g., '0-1').

### Request Example
```python
from esorm import setup_mappings, connect
from esorm.model import create_index_template, set_default_index_prefix

async def prepare_elasticsearch():
    # Connect to ES
    await connect('localhost:9200', wait=True)

    # Optional: set custom index prefix (default is 'esorm_')
    set_default_index_prefix('myapp_')

    # Create index template
    await create_index_template(
        'default_template',
        prefix_name='myapp_',
        shards=3,
        auto_expand_replicas='1-5'
    )

    # Import models first
    import models

    # Create indices and mappings for all registered models
    await setup_mappings(models)
```
```

--------------------------------

### Setup Index Mappings with ESORM

Source: https://context7.com/wallneradam/esorm/llms.txt

Provides a Python function to prepare ElasticSearch indices and mappings using ESORM. It covers connecting to the cluster, setting index prefixes, creating index templates, and then applying mappings to registered models.

```python
from esorm import setup_mappings, connect
from esorm.model import create_index_template, set_default_index_prefix

async def prepare_elasticsearch():
    # Connect to ES
    await connect('localhost:9200', wait=True)

    # Optional: set custom index prefix (default is 'esorm_')
    set_default_index_prefix('myapp_')

    # Create index template
    await create_index_template(
        'default_template',
        prefix_name='myapp_',
        shards=3,
        auto_expand_replicas='1-5'
    )

    # Import models first
    import models

    # Create indices and mappings for all registered models
    await setup_mappings(models)
```

--------------------------------

### Connect to ElasticSearch with Connection Strings

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Provides examples of establishing a connection to ElasticSearch using the `esorm.connect` function. It covers connecting to a single host, multiple hosts for a cluster, and options for waiting until the connection is ready (`wait=True`) and configuring connection details like `sniff_on_start` and `sniff_on_connection_fail`.

```python
from esorm import connect


async def es_init():
    await connect('localhost:9200')
```

```python
from esorm import connect


async def es_init():
    await connect(['localhost:9200', 'localhost:9201'])
```

```python
from esorm import connect


async def es_init():
    await connect('localhost:9200', wait=True)
```

```python
from esorm import connect


async def es_init():
    await connect('localhost:9200', wait=True, sniff_on_start=True, sniff_on_connection_fail=True)
```

--------------------------------

### Install ESORM Package

Source: https://github.com/wallneradam/esorm/blob/main/README.md

This command installs the ESORM Python package using pip. ESORM is a library that provides an Object Document Mapper for ElasticSearch, built on top of Pydantic, offering asynchronous support and automatic mapping capabilities.

```bash
pip install pyesorm
```

--------------------------------

### Python Bulk Operations with ESORM

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Demonstrates how to perform bulk create and delete operations using ESORM's ESBulk class. It includes examples for immediate refresh and waiting for scheduled refreshes. Dependencies include the 'esorm' library.

```python
from esorm import ESModel, ESBulk
from typing import List

class User(ESModel):
    name: str
    age: int

async def bulk_create_users():
    async with ESBulk() as bulk:
        # Creating or modifiying models
        for i in range(10):
            user = User(name=f'User {i}', age=i)
            await bulk.save(user)

async def bulk_delete_users(users: List[User]):
    async with ESBulk(wait_for=True) as bulk:  # Wait for scheduled refresh
        # Deleting models
        for user in users:
            await bulk.delete(user)

async def bulk_create_users_immediate():
    async with ESBulk(refresh=True) as bulk:  # Force immediate refresh
        for i in range(10):
            user = User(name=f'User {i}', age=i)
            await bulk.save(user)
```

--------------------------------

### Python - ESORM Shard Routing

Source: https://context7.com/wallneradam/esorm/llms.txt

Demonstrates how to implement shard routing in ESORM for colocalizing data, using the `__routing__` property and passing the routing key during save and search operations. This example routes comments by their associated post ID.

```python
class Comment(ESModel):
    id: int
    post_id: int
    content: str

    @property
    def __routing__(self) -> str:
        """Route by post_id to colocate comments with posts"""
        return str(self.post_id)

async def routing_example():
    comment = Comment(id=1, post_id=42, content='Great post!')
    await comment.save(routing='42')

    # Search with routing
    comments = await Comment.search(
        {'match': {'post_id': 42}},
        routing='42'
    )
```

--------------------------------

### Handle Elasticsearch Errors in Python

Source: https://context7.com/wallneradam/esorm/llms.txt

Provides examples of how to handle common Elasticsearch errors when using ESORM, such as NotFoundError, BulkError, and ConflictError, using try-except blocks.

```python
from esorm.error import NotFoundError, InvalidResponseError, BulkError, ConflictError

async def error_handling_examples():
    # Handle not found
    try:
        user = await User.get('nonexistent_id')
    except NotFoundError as e:
        print(f"User not found: {e}")

    # Handle bulk errors
    try:
        async with ESBulk() as bulk:
            # ... bulk operations
            pass
    except BulkError as e:
        for error in e.errors:
            print(f"Status: {error['status']}")
            print(f"Type: {error['type']}")
            print(f"Reason: {error['reason']}")
            print(f"Model: {error['model']}")

    # Handle version conflicts
    try:
        await user.save()
    except ConflictError as e:
        print("Document was modified, reload and retry")
```

--------------------------------

### Describe Fields in ESModel using Docstrings and Pydantic

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Demonstrates two methods for describing fields in an `esorm.ESModel`: using docstrings directly below the field definition for intuitive readability, or using the standard Pydantic `TextField` with a description. This is useful for generating API documentation, for example, with FastAPI.

```python
from esorm import ESModel
from esorm.fields import TextField


class User(ESModel):
    name: str = 'John Doe'
    """ The name of the user """
    age: int = 18
    """ The age of the user """

    # This is the usual Pydantic way, but I think docstrings are more intuitive and readable
    address: str = TextField(description="The address of the user")
```

--------------------------------

### Python kNN Vector Search with ESORM

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Provides an example of performing a k-nearest neighbors (kNN) vector search using ESORM. This enables semantic search and similarity operations. Requires the 'esorm' library.

```python
# Vector search using kNN
results = await Document.search({
    "knn": {
        "field": "embedding",
        "query_vector": [0.1, 0.2, ...],  # your query vector
        "k": 10,  # number of neighbors to return
        "num_candidates": 100  # number of candidates to consider
    }
})
```

--------------------------------

### Setup ESORM Mappings and Create Indices

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

This snippet illustrates how to automatically create Elasticsearch indices and their mappings based on defined ESORM model classes. It requires importing all model classes first, which are registered globally. The `setup_mappings` function then processes these models to create the necessary Elasticsearch structures. Note that this method ignores mapping errors for existing indices and can add new fields but cannot modify or delete existing ones without reindexing.

```python
import asyncio
from esorm import setup_mappings

async def prepare_es():
    import models  # Import your models here
    await setup_mappings(models)

# To run the async function:
# asyncio.run(prepare_es())
```

--------------------------------

### Combine Keyword Shortcut with Custom Fields in Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Illustrates how to leverage both the `keyword=True` shortcut and custom fields simultaneously when defining esorm models. The `keyword` subfield is automatically added if not explicitly defined. This example also shows how custom configurations for a `keyword` field take precedence, ensuring specific settings like `ignore_above` are preserved.

```python
from esorm import ESModel
from esorm.fields import TextField

class Product(ESModel):
    # Both keyword and custom fields
    title: str = TextField(..., keyword=True, fields={
        'suggest': {'type': 'completion'},
        'ngram': {'type': 'text', 'analyzer': 'ngram'}
    })
    # Results in: {
    #     'keyword': {'type': 'keyword'},  # Added automatically
    #     'suggest': {'type': 'completion'},
    #     'ngram': {'type': 'text', 'analyzer': 'ngram'}
    # }

    # Custom keyword configuration takes precedence
    name: str = TextField(..., keyword=True, fields={
        'keyword': {'type': 'keyword', 'ignore_above': 512}  # Custom config preserved
    })
```

--------------------------------

### Python - ESORM Optimistic Concurrency Control

Source: https://context7.com/wallneradam/esorm/llms.txt

Shows how to implement optimistic concurrency control using the `@retry_on_conflict` decorator in ESORM. Examples include safe updates with a fixed number of retries and infinite retries with document reloading.

```python
from esorm import retry_on_conflict

@retry_on_conflict(max_retries=5)
async def update_user_safe(user_id: str):
    user = await User.get(user_id)
    user.age += 1
    await user.save()  # Retries on conflict

@retry_on_conflict(max_retries=-1, reload_on_conflict=True)
async def update_with_reload(user_id: str):
    user = await User.get(user_id)
    user.age += 1
    await user.save()  # Infinite retries, reloads document on conflict
```

--------------------------------

### Python Aggregations: Average Age and Count by Country

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

Shows how to use the `aggregate` method in Python to perform aggregations on ES data. It covers calculating the average age of users and the number of users per country. The examples demonstrate defining aggregation queries and optionally filtering the documents to aggregate using a 'match' query. Type checking and IDE autocompletion are supported due to `TypedDict` annotations.

```python
from esorm import ESModel

class User(ESModel):
    name: str
    age: int
    country: str

async def aggregate_avg():
    # Get average age of users
    aggs_def = {
        'avg_age': {
            'avg': {
                'field': 'age'
            }
        }
    }
    aggs = await User.aggregate(aggs_def)
    print(aggs['avg_age']['value'])

async def aggregate_avg_by_country(country = 'Hungary'):
    # Get average age of users by country
    aggs_def = {
        'avg_age': {
            'avg': {
                'field': 'age'
            }
        }
    }
    query = {
        'bool': {
            'must': [{
                'match': {
                    'country': {
                        'query': country
                    }
                }
            }]
        }
    }
    aggs = await User.aggregate(aggs_def, query)
    print(aggs['avg_age']['value'])


async def aggregate_terms():
    # Get number of users by country
    aggs_def = {
        'countries': {
            'terms': {
                'field': 'country'
            }
        }
    }
    aggs = await User.aggregate(aggs_def)
    for bucket in aggs['countries']['buckets']:
        print(bucket['key'], bucket['doc_count'])
```

--------------------------------

### Python: Search Documents by Specific Fields

Source: https://context7.com/wallneradam/esorm/llms.txt

Provides examples of searching for documents based on specific field values. Includes searching by multiple fields, paginating field searches, finding a single document by fields, and retrieving all documents in a collection.

```python
async def search_by_fields():
    # Simple field search
    users = await User.search_by_fields({'age': 18, 'country': 'USA'})

    # With pagination
    users = await User.search_by_fields(
        {'country': 'Hungary'},
        page=1,
        page_size=50
    )

    # Search one by fields
    user = await User.search_one_by_fields({'email': 'john@example.com'})

    # Get all documents
    all_users = await User.all()
```

--------------------------------

### Retrieve Selected Fields Only with ESORM (Python)

Source: https://github.com/wallneradam/esorm/blob/main/docs/advanced.md

Provides examples of how to retrieve only specific fields from Elasticsearch documents using the `_source` argument in ESORM's `get` and `search_one_by_fields` methods. This technique requires default values to be set for all fields in the model to handle cases where fields are not returned.

```python
import esorm

class Model(esorm.ESModel):
    f_int: int = 0
    f_str: str = 'a'


async def test_source():    
    doc = Model(f_int=1, f_str='b')
    doc_id = await doc.save()

    doc = await Model.get(doc_id)
    assert doc.f_str == 'b'
    assert doc.f_int == 1

    doc = await Model.search_one_by_fields(dict(_id=doc_id), _source=['f_str'])
    assert doc.f_str == 'b'
    assert doc.f_int == 0

    doc = await Model.search_one_by_fields(dict(_id=doc_id), _source=['f_int'])
    assert doc.f_str == 'a'
    assert doc.f_int == 1

    doc = await Model.get(doc_id, _source=['f_str'])
    assert doc.f_str == 'b'
    assert doc.f_int == 0

    doc = await Model.get(doc_id, _source=['f_int'])
    assert doc.f_str == 'a'
    assert doc.f_int == 1
```

--------------------------------

### Control Refresh Behavior for ESORM Save Operations

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

This example showcases different ways to control when changes made via `user.save()` become visible in Elasticsearch searches. Options include default behavior (no refresh), immediate refresh (`refresh=True`), waiting for the next refresh cycle (`wait_for=True` or `refresh='wait_for'`). The default offers the best performance, while `refresh=True` is suitable for testing, and `wait_for` provides a balance.

```python
# Assuming 'user' is an existing ESModel instance

# Default: No refresh - fastest, changes visible after refresh_interval
# await user.save()

# Immediate refresh - changes immediately visible to search
# await user.save(refresh=True)

# Wait for next refresh cycle - waits for the scheduled refresh_interval
# await user.save(wait_for=True)

# Alternative: ES API compatible syntax (same as wait_for=True)
# await user.save(refresh="wait_for")
```

--------------------------------

### Implement Shard Routing in ESORM (Python)

Source: https://github.com/wallneradam/esorm/blob/main/docs/advanced.md

Demonstrates how to implement shard routing for an ESORM model by defining a `__routing__` property. This property returns a routing value that Elasticsearch uses to direct documents to specific shards, improving query performance. An example async function `get_user_by_region` shows how to use shard routing in a search query.

```python
from typing import List
from esorm import ESModel

class User(ESModel):
    first_name: str
    last_name: str
    region: str
    
    @property
    def __routing__(self) -> str:
        """ Return the routing value for this document """
        return self.region + '_routing'  # Calculate the routing value from the region field
        
async def get_user_by_region(region: str = 'europe') -> List[User]:
    """ Search for users by region using shard routing """
    return await User.search_by_fields(region=region, routing=f"{region}_routing")
```

--------------------------------

### Automatic Conflict Retries with retry_on_conflict Decorator in Python

Source: https://github.com/wallneradam/esorm/blob/main/docs/advanced.md

Shows how to use the `retry_on_conflict` decorator in Python with ESORM to automatically retry operations that encounter optimistic concurrency control conflicts. This example demonstrates retrying a function that increments a user's login count.

```python
import asyncio
from esorm import ESModel, retry_on_conflict


class User(ESModel):
    first_name: str
    last_name: str
    logins: int = 0


async def test_retry_on_conflict(user: User):    
    @retry_on_conflict(3)  # Retry 3 times on conflict
    async def login(user_id):
        _user = await User.get(id=user_id)
        _user.logins += 1
    
    # This won't raise a ConflictError   
    await asyncio.gather(
        login(user._id),
        login(user._id),
        login(user._id),
    )
```

--------------------------------

### Control Refresh Behavior on Save using Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

This example illustrates different ways to control when changes made via the `save` method become visible to search operations in Elasticsearch. Options include default behavior (no refresh), immediate refresh (`refresh=True`), waiting for the next refresh cycle (`wait_for=True`), or using Elasticsearch API compatible syntax (`refresh='wait_for'`).

```python
# Default: No refresh - fastest, changes visible after refresh_interval
await user.save()

# Immediate refresh - changes immediately visible to search
await user.save(refresh=True)

# Wait for next refresh cycle - waits for the scheduled refresh_interval
await user.save(wait_for=True)

# Alternative: ES API compatible syntax (same as wait_for=True)
await user.save(refresh="wait_for")
```

--------------------------------

### Retrieve a User Document by ID using ESORM

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

This code snippet demonstrates how to fetch a specific document from Elasticsearch using the `get` method provided by ESORM. It requires the document's ID as input. The method returns an instance of the model class populated with the data from Elasticsearch, including internal Elasticsearch attributes like `_id`, `_routing`, etc.

```python
import asyncio
from esorm import ESModel

class User(ESModel):
    name: str
    age: int

async def get_user(user_id: str):
    user = await User.get(user_id)
    if user:
        print(f"User name: {user.name}")
    else:
        print(f"User with ID {user_id} not found.")

# Example usage:
# asyncio.run(get_user('some_user_id'))
```

--------------------------------

### Use List of Primitive Fields in Python with esorm

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Illustrates how to define fields that are lists of primitive data types (like strings or integers) in esorm models. The `User` model example shows `emails` as a list of strings and `favorite_ids` as a list of integers, which is a common requirement for storing multiple values for a single attribute.

```python
from typing import List
from esorm import ESModel


class User(ESModel):
    emails: List[str]
    favorite_ids: List[int]
    ...
```

--------------------------------

### Update a User Document using Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

This example demonstrates how to update an existing user document in Elasticsearch. It first retrieves the user by ID, modifies one of its attributes (e.g., `name`), and then calls `save()` on the modified user object. ESORM automatically handles optimistic locking using `_primary_term` and `_seq_no` fields during the update process.

```python
from esorm import ESModel


# Here the model have automatically generated id
class User(ESModel):
    name: str
    age: int


async def update_user(user_id: str):
    user = await User.get(user_id)
    user.name = 'Jane Doe'
    await user.save()
```

--------------------------------

### Define Nested Documents in Python with esorm

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Explains how to define nested documents within esorm models. The example shows a `User` model defined with basic field types and a `Post` model that includes a `writer` field typed as `User`. This structure allows for complex, hierarchical data representation within Elasticsearch.

```python
from esorm import ESModel
from esorm.fields import keyword, text, byte


class User(ESModel):
    name: text
    email: keyword
    age: byte = 18


class Post(ESModel):
    title: text
    content: text
    writer: User  # User is a nested document
```

--------------------------------

### Update a User Document in ESORM

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

This example shows how to update an existing document in Elasticsearch using ESORM. First, it retrieves the document using `User.get(user_id)`. Then, it modifies the desired fields (e.g., `user.name`). Finally, calling `user.save()` again persists the changes. ESORM automatically handles optimistic locking using `_primary_term` and `_seq_no` to prevent race conditions during updates.

```python
import asyncio
from esorm import ESModel

class User(ESModel):
    name: str
    age: int

async def update_user(user_id: str):
    user = await User.get(user_id)
    if user:
        user.name = 'Jane Doe'
        await user.save()
        print(f"User {user_id} updated successfully.")
    else:
        print(f"User with ID {user_id} not found.")

# Example usage:
# asyncio.run(update_user('some_user_id'))
```

--------------------------------

### Utilize Dense Vector Field for Vector Similarity Search in Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Demonstrates the use of the `dense_vector` field type for enabling vector similarity search in Elasticsearch (ES 8.x and later). The example defines a `Document` model with an `embedding` field, specifying its dimensions and the similarity metric (cosine, dot_product, or l2). This field type is crucial for applications involving machine learning embeddings and similarity-based retrieval.

```python
from esorm import ESModel
from esorm.fields import dense_vector, DenseVectorField

class Document(ESModel):
    id: str
    content: str
    embedding: dense_vector = DenseVectorField(
        ...,  # required field
        dims=384,  # dimension of the vector
        similarity="cosine"  # similarity metric: 'cosine', 'dot_product', or 'l2'
    )
```

--------------------------------

### Bulk Create Users (Immediate Refresh)

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

Demonstrates bulk saving with `refresh=True`, forcing an immediate refresh after operations. Useful for testing or when immediate visibility is critical.

```python
from esorm import ESModel, ESBulk
from typing import List


class User(ESModel):
    name: str
    age: int


async def bulk_create_users_immediate():
    async with ESBulk(refresh=True) as bulk:  # Force immediate refresh
        for i in range(10):
            user = User(name=f'User {i}', age=i)
            await bulk.save(user)
```

--------------------------------

### Connect to ElasticSearch with ESORM

Source: https://context7.com/wallneradam/esorm/llms.txt

Demonstrates various ways to establish a connection to an ElasticSearch cluster using the `esorm.connect` function. Supports single host, multiple hosts, waiting for readiness, and advanced connection options.

```python
from esorm import connect

# Basic connection
await connect('localhost:9200')

# Multiple hosts (cluster)
await connect(['localhost:9200', 'localhost:9201'])

# Wait for ElasticSearch to be ready
await connect('localhost:9200', wait=True)

# Advanced options
await connect('localhost:9200', wait=True, sniff_on_start=True, sniff_on_connection_fail=True)
```

--------------------------------

### Bulk Create Users (Default Refresh)

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

Demonstrates how to perform a bulk save of User models using ESBulk. Changes are visible after the index's refresh_interval.

```python
from esorm import ESModel, ESBulk
from typing import List


class User(ESModel):
    name: str
    age: int


async def bulk_create_users():
    async with ESBulk() as bulk:
        # Creating or modifiying models
        for i in range(10):
            user = User(name=f'User {i}', age=i)
            await bulk.save(user)
```

--------------------------------

### Connect to ElasticSearch

Source: https://context7.com/wallneradam/esorm/llms.txt

Establishes a connection to the ElasticSearch cluster using various configurations.

```APIDOC
## Connect to ElasticSearch

### Description
Establishes a connection to the ElasticSearch cluster using various configurations.

### Method
`connect` function

### Endpoint
N/A (Client-side connection)

### Parameters
#### Arguments
- **hosts** (str or list[str]) - Required - The host(s) to connect to (e.g., 'localhost:9200' or ['host1:9200', 'host2:9201']).
- **wait** (bool) - Optional - If True, waits for ElasticSearch to be ready before returning.
- **sniff_on_start** (bool) - Optional - If True, sniffs cluster nodes on connection start.
- **sniff_on_connection_fail** (bool) - Optional - If True, sniffs cluster nodes when a connection fails.

### Request Example
```python
from esorm import connect

# Basic connection
await connect('localhost:9200')

# Multiple hosts (cluster)
await connect(['localhost:9200', 'localhost:9201'])

# Wait for ElasticSearch to be ready
await connect('localhost:9200', wait=True)

# Advanced options
await connect('localhost:9200', wait=True, sniff_on_start=True, sniff_on_connection_fail=True)
```
```

--------------------------------

### FastAPI Pagination and Sorting with ESORM (Python)

Source: https://github.com/wallneradam/esorm/blob/main/docs/advanced.md

This Python code illustrates advanced FastAPI integration with ESORM, specifically focusing on implementing pagination and sorting for API endpoints. It uses FastAPI's dependency injection system with helper functions from `esorm.fastapi` to manage query parameters for pagination and sorting. The endpoint automatically sets the `X-Total-Hits` header. Dependencies include `typing`, `esorm`, `fastapi`, and `esorm.fastapi`.

```python
from typing import List
from esorm import ESModelTimestamp, Pagination, Sort
from fastapi import FastAPI, Depends

from esorm.fastapi import make_dep_sort, make_dep_pagination


class User(ESModelTimestamp):
    """ The User model """
    first_name: str
    last_name: str
    

app = FastAPI()

@app.get("/all_users")
async def all_users(
    # This will create a _page, and a _page_size query parameter for the endpoint 
    pagination: Pagination = Depends(make_dep_pagination(default_page=1, default_page_size=10)),
    # This will create a _sort enum query parameter for the endpoint, so it is selectable in swagger UI
    sort: Sort = Depends(make_dep_sort( 
        first_name_last_name_asc=  # This is the name of the 1st sort option
        # Definition of the sort options
        [
            {'first_name': {"order": "asc"}},
            {'last_name': {"order": "asc"}},
        ],
        last_name_first_name_asc=  # This is the name of the 2nd sort option
        # Definition of the sort options
        [
            {'last_name': {"order": "asc"}},
            {'first_name': {"order": "asc"}},
        ]
    )),
) -> List[User]:
    """ Get all users """
    return await sort(pagination(User)).all()
```

--------------------------------

### Python: Perform Bulk Operations with ESBulk

Source: https://context7.com/wallneradam/esorm/llms.txt

Demonstrates how to perform bulk create, update, and delete operations using the ESBulk context manager. Supports default asynchronous saving, immediate refresh, waiting for completion, and optimistic concurrency control for updates.

```python
from esorm import ESBulk, ESModel
from typing import List

class User(ESModel):
    name: str
    age: int

async def bulk_create_users():
    # Default: no refresh
    async with ESBulk() as bulk:
        for i in range(100):
            user = User(name=f'User {i}', age=i)
            await bulk.save(user)
    # All users saved in single bulk request

async def bulk_with_immediate_refresh():
    # Force immediate refresh
    async with ESBulk(refresh=True) as bulk:
        for i in range(10):
            user = User(name=f'User {i}', age=i)
            await bulk.save(user)

async def bulk_delete_users(users: List[User]):
    async with ESBulk(wait_for=True) as bulk:
        for user in users:
            await bulk.delete(user)

# Bulk operations support optimistic concurrency control
async def bulk_update_existing():
    users = await User.search({'match_all': {}})
    async with ESBulk() as bulk:
        for user in users:
            user.age += 1
            await bulk.save(user)  # Will use _seq_no and _primary_term
```

--------------------------------

### Python - ESORM Pagination and Sorting

Source: https://context7.com/wallneradam/esorm/llms.txt

Illustrates how to implement pagination and sorting for search results using ESORM's Pagination and Sort classes. It shows basic usage, combined with field searches, and simplified sorting syntax.

```python
from esorm.model import Pagination, Sort

class User(ESModel):
    id: int
    name: str
    age: int

async def paginated_search():
    # Using Pagination decorator
    def pagination_callback(total: int):
        print(f'Total users: {total}')

    pagination = Pagination(page=2, page_size=20, callback=pagination_callback)
    users = await pagination(User).search_by_fields({'age': 18})

    # Using Sort decorator
    sort = Sort(sort=[
        {'age': {'order': 'desc'}},
        {'name': {'order': 'asc'}}
    ])
    users = await sort(User).search_by_fields({'age': 18})

    # Simplified sort syntax
    sort = Sort(sort='name')
    users = await sort(User).all()

    # Combine pagination and sorting
    users = await pagination(sort(User)).search({'match_all': {}})
```

--------------------------------

### FastAPI Integration with ESORM Models (Python)

Source: https://github.com/wallneradam/esorm/blob/main/docs/advanced.md

This Python code demonstrates how to integrate ESORM models with FastAPI. It defines a `User` model inheriting from `ESModelTimestamp` and creates API endpoints for creating and searching users. FastAPI handles request parsing and response serialization. Dependencies include `typing`, `esorm`, and `fastapi`.

```python
from typing import List, Optional
from esorm import ESModelTimestamp
from fastapi import FastAPI


class User(ESModelTimestamp):
    """ The User model """
    first_name: str
    last_name: str
    

app = FastAPI()


@app.post("/users")
async def create_user(first_name: str, last_name: str) -> User:
    """ Create a new user """
    user = User(first_name=first_name, last_name=last_name)
    await user.save()
    return user


@app.get("/users")
async def users(first_name: Optional[str] = None, last_name: Optional[str] = None) -> List[User]:
    """ Search users """
    return await User.search_by_fields(first_name=first_name, last_name=last_name)
```

--------------------------------

### Create Index Template with ESORM

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

This snippet demonstrates how to create an index template in Elasticsearch using the ESORM library. It specifies the template name, a prefix for indices, and the number of primary shards and replica configurations. This is useful for defining default settings for indices managed by ESORM.

```python
import asyncio
from esorm.model import ESModel

async def prepare_es():
    await ESModel.create_index_template('default_template',
                                            prefix_name='esorm_',
                                            shards=3,
                                            auto_expand_replicas='1-5')

# To run the async function:
# asyncio.run(prepare_es())
```

--------------------------------

### FastAPI Integration for Pagination and Sorting in Python

Source: https://context7.com/wallneradam/esorm/llms.txt

Shows how to integrate ESORM with FastAPI using utility functions to create reusable dependencies for pagination and sorting. This simplifies handling dynamic query parameters in API endpoints.

```python
from fastapi import FastAPI, Depends
from esorm.fastapi import make_dep_pagination, make_dep_sort, set_max_page_size

app = FastAPI()

# Set global max page size
set_max_page_size(1000)

# Create reusable dependencies
dep_pagination = make_dep_pagination(default_page=1, default_page_size=20)
dep_sort = make_dep_sort(
    by_age=[{'age': {'order': 'desc'}}],
    by_name=[{'name.keyword': {'order': 'asc'}}]
)

@app.get("/users")
async def list_users(
    pagination: Pagination = Depends(dep_pagination),
    sort: Sort = Depends(dep_sort)
):
    users = await pagination(sort(User)).search({'match_all': {}})
    return users

# Query params: ?_page=2&_page_size=50&_sort=by_age
# Response includes X-Total-Hits header
```

--------------------------------

### Create Index Template using Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

This snippet demonstrates how to create an Elasticsearch index template with a specified name, shard count, and replica configuration using the esorm library. It ensures that all indices prefixed with 'esorm_' (or a custom prefix) adhere to these template settings.

```python
async def prepare_es():
    await esorm_model.create_index_template('default_template',
                                            prefix_name='esorm_',
                                            shards=3,
                                            auto_expand_replicas='1-5')
```

--------------------------------

### ESORM CRUD: Create Documents

Source: https://context7.com/wallneradam/esorm/llms.txt

Demonstrates how to create and save new documents in ElasticSearch using ESORM. Covers auto-generated IDs, forcing immediate refresh, and waiting for refresh operations.

```python
from esorm import ESModel

class User(ESModel):
    name: str
    age: int
    email: str

async def create_user():
    # Create and save with auto-generated ID
    user = User(name='John Doe', age=25, email='john@example.com')
    new_id = await user.save()
    print(f"Created user with ID: {new_id}")

    # Force immediate refresh (visible to search immediately)
    await user.save(refresh=True)

    # Wait for scheduled refresh
    await user.save(wait_for=True)
    # Or ES API compatible syntax
    await user.save(refresh="wait_for")
```

--------------------------------

### Python General Search with ESORM

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Shows how to perform general searches and retrieve a single document using ESORM's search and search_one methods. It supports complex queries and returning results as dictionaries. Requires the 'esorm' library.

```python
from esorm import ESModel

class User(ESModel):
    name: str
    age: int

async def search_users():
    # Search for users at least 18 years old
    users = await User.search(
        query={
            'bool': {
                'must': [{
                    'range': {
                        'age': {
                            'gte': 18
                        }
                    }
                }]
            }
        }
    )
    for user in users:
        print(user.name)

async def search_one_user():
    # Search a user named John Doe
    user = await User.search_one(
        query={
            'bool': {
                'must': [{
                    'match': {
                        'name': {
                            'query': 'John Doe'
                        }
                    }
                }]
            }
        }
    )
    print(user.name)
```

--------------------------------

### ESORM Bulk Operations Context Manager

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

This snippet introduces the `ESBulk` context manager for performing bulk operations (create, update, delete) efficiently in Elasticsearch with ESORM. Using a context manager can significantly improve performance when dealing with a large number of documents compared to individual operations.

```python
from typing import List
from esorm import ESModel, ESBulk

# Example model
class MyModel(ESModel):
    field1: str

async def bulk_operation_example():
    # Assuming 'documents_to_process' is a list of documents
    # For example: documents_to_process = [MyModel(field1='data1'), MyModel(field1='data2')]
    
    async with ESBulk() as bulk:
        # Add documents to the bulk operation
        # for doc in documents_to_process:
        #     await bulk.save(doc)
        #     await bulk.delete(doc_id='some_id')
        pass # Replace with actual bulk operations

# To run the async function:
# asyncio.run(bulk_operation_example())
```

--------------------------------

### Bulk Delete Users (Wait for Refresh)

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

Shows how to perform a bulk delete of User models with `wait_for=True`, ensuring operations complete before proceeding. Good balance between consistency and performance.

```python
from esorm import ESModel, ESBulk
from typing import List


class User(ESModel):
    name: str
    age: int


async def bulk_delete_users(users: List[User]):
    async with ESBulk(wait_for=True) as bulk:  # Wait for scheduled refresh
        # Deleting models
        for user in users:
            await bulk.delete(user)
```

--------------------------------

### Python: Execute Searches with Query DSL

Source: https://context7.com/wallneradam/esorm/llms.txt

Shows how to perform various searches using Elasticsearch Query DSL dictionaries. Includes searching with boolean queries, retrieving results as a dictionary keyed by ID, finding a single document, and paginating/sorting results.

```python
from esorm import ESModel

class User(ESModel):
    name: str
    age: int
    country: str

async def search_users():
    # Search with query dict (type-checked)
    users = await User.search({
        'bool': {
            'must': [{
                'range': {
                    'age': {'gte': 18}
                }
            }]
        }
    })

    # Get results as dict keyed by ID
    users_dict = await User.search(
        query={'match_all': {}},
        res_dict=True
    )

    # Search one result
    user = await User.search_one({
        'bool': {
            'must': [{
                'match': {
                    'name': {'query': 'John Doe'}
                }
            }]
        }
    })

    # Search with pagination and sorting
    users = await User.search(
        query={'match_all': {}},
        page=2,
        page_size=20,
        sort=[{'age': {'order': 'desc'}}]
    )

    # Or simplified sort syntax
    users = await User.search(
        query={'match_all': {}},
        sort='age'  # Defaults to ascending
    )
```

--------------------------------

### Define Field Descriptions in Python

Source: https://context7.com/wallneradam/esorm/llms.txt

Illustrates two methods for defining descriptions for Elasticsearch fields within an ESORM model: using docstrings (recommended) and Pydantic's description argument.

```python
from esorm import ESModel
from esorm.fields import TextField

class User(ESModel):
    # Method 1: Docstring (recommended)
    name: str = 'John Doe'
    """ The name of the user """

    age: int = 18
    """ The age of the user """

    # Method 2: Pydantic description
    address: str = TextField(description="The address of the user")
```

--------------------------------

### Pagination Decorator for ESORM Models

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Shows how to implement pagination for search results using the `Pagination` class. This snippet defines a `User` model and then applies the `Pagination` decorator to control the page number and page size of the search query, optionally invoking a callback with the total number of documents found.

```python
from esorm.model import ESModel, Pagination


class User(ESModel):
    id: int  # This will be used as the document _id in the index
    name: str
    age: int


def get_users(page = 1, page_size = 10):

    def pagination_callback(total: int):
        # You may set a header value or something else here
        print(f'Total users: {total}')

    # 1st create the decorator itself
    pagination = Pagination(page=page, page_size=page_size)

    # Then decorate your model
    res = pagination(User).search_by_fields(age=18)

    # Here the result has maximum 10 items
    return res
```

--------------------------------

### ESORM Specialized Field Definition

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Shows how to define fields with specialized configurations using ESORM's field definition functions. This allows for more control over field properties like minimum/maximum length or range constraints.

```python
from esorm.fields import Field, TextField, NumericField, DenseVectorField

class Product(ESModel):
    id: str
    name: str = TextField(..., min_length=3, max_length=100)
    price: float = NumericField(..., gt=0)
    is_available: bool = Field(True)
    location: geo_point
    embedding: dense_vector = DenseVectorField(..., dims=384, similarity="cosine")
```

--------------------------------

### Python Document Count with ESORM

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Demonstrates how to count documents in an index using ESORM's count method, with options for filtering by a query. Requires the 'esorm' library.

```python
from esorm import ESModel

class User(ESModel):
    name: str
    age: int

async def count_users():
    count = await User.count()
    print(count)

async def count_users_by_age():
    count = await User.count(query={'age': 18})
    print(count)
```

--------------------------------

### Create Indices and Mappings from Models using Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

This Python function automatically generates and creates Elasticsearch indices and their mappings based on defined model classes. It first registers all model classes and then calls `setup_mappings` to create the necessary structures in Elasticsearch. Note that this method ignores mapping errors for existing indices and can update indices with new fields but cannot modify or delete fields without reindexing.

```python
from esorm import setup_mappings

async def prepare_es():
    import models  # Import your models
    # Here models argument is not needed, but you can pass it to prevent unused import warning
    await setup_mappings(models)
```

--------------------------------

### Python Field-Value Search with ESORM

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Illustrates searching for documents based on specific field values using ESORM's search_by_fields and search_one_by_fields methods. Supports returning results as dictionaries. Requires the 'esorm' library.

```python
from esorm import ESModel

class User(ESModel):
    name: str
    age: int

async def search_users():
    # Search users age is 18
    users = await User.search_by_fields({'age': 18})
    for user in users:
        print(user.name)
```

--------------------------------

### ESORM Python Basic Types Mapping

Source: https://github.com/wallneradam/esorm/blob/main/README.md

Demonstrates how basic Python types are mapped to Elasticsearch types using ESORM. It shows the direct conversion for common types like str, int, float, bool, and datetime objects.

```python
from esorm import ESModel


class User(ESModel):
    name: str
    age: int
```

--------------------------------

### Combine Keyword Shortcut with Custom Fields in ESORM

Source: https://github.com/wallneradam/esorm/blob/main/docs/README.md

Illustrates how to use the 'keyword=True' shortcut alongside custom 'fields' in ESORM. The 'keyword' subfield is automatically added if not present, and custom configurations for 'keyword' fields take precedence.

```python
from esorm import ESModel
from esorm.fields import TextField

class Product(ESModel):
    # Both keyword and custom fields
    title: str = TextField(..., keyword=True, fields={
        'suggest': {'type': 'completion'},
        'ngram': {'type': 'text', 'analyzer': 'ngram'}
    })
    # Results in: {
    #     'keyword': {'type': 'keyword'},  # Added automatically
    #     'suggest': {'type': 'completion'},
    #     'ngram': {'type': 'text', 'analyzer': 'ngram'}
    # }
    
    # Custom keyword configuration takes precedence
    name: str = TextField(..., keyword=True, fields={
        'keyword': {'type': 'keyword', 'ignore_above': 512}  # Custom config preserved
    })
```

--------------------------------

### Python - ESORM Aggregations

Source: https://context7.com/wallneradam/esorm/llms.txt

Demonstrates how to perform various aggregations (average, terms, filtered average, multi-metric) using the ESORM library in Python. It requires an ESModel class (e.g., User) defined with relevant fields.

```python
class User(ESModel):
    name: str
    age: int
    country: str
    salary: float

async def aggregate_examples():
    # Average age
    aggs_def = {
        'avg_age': {
            'avg': {'field': 'age'}
        }
    }
    result = await User.aggregate(aggs_def)
    print(f"Average age: {result['avg_age']['value']}")

    # Terms aggregation
    aggs_def = {
        'countries': {
            'terms': {'field': 'country', 'size': 10}
        }
    }
    result = await User.aggregate(aggs_def)
    for bucket in result['countries']['buckets']:
        print(f"{bucket['key']}: {bucket['doc_count']}")

    # Aggregation with query filter
    aggs_def = {
        'avg_salary': {
            'avg': {'field': 'salary'}
        }
    }
    query = {
        'bool': {
            'must': [{'match': {'country': {'query': 'USA'}}}]
        }
    }
    result = await User.aggregate(aggs_def, query=query)
    print(f"Average USA salary: {result['avg_salary']['value']}")

    # Complex multi-metric aggregation
    aggs_def = {
        'by_country': {
            'terms': {'field': 'country'},
            'aggs': {
                'avg_age': {'avg': {'field': 'age'}},
                'max_salary': {'max': {'field': 'salary'}}
            }
        }
    }
    result = await User.aggregate(aggs_def)
```

--------------------------------

### Python: Count Documents with Various Criteria

Source: https://context7.com/wallneradam/esorm/llms.txt

Illustrates how to count documents in Elasticsearch using different methods. Includes counting all documents, counting based on a query, and counting documents that match specific field values.

```python
async def count_examples():
    # Count all documents
    total = await User.count()

    # Count with query
    adults = await User.count({
        'bool': {
            'must': [{'range': {'age': {'gte': 18}}}]
        }
    })

    # Count by fields
    us_users = await User.count({'country': 'USA'})
```

--------------------------------

### Define and Save a User Model using Python

Source: https://github.com/wallneradam/esorm/blob/main/README.md

This snippet defines a `User` model inheriting from `ESModel` with `name` and `age` fields. It then demonstrates how to create a new user instance, save it to Elasticsearch, and retrieve the generated document ID. The `save` method supports refresh control options.

```python
from esorm import ESModel


# Here the model have automatically generated id
class User(ESModel):
    name: str
    age: int


async def create_user():
    # Create a new user
    user = User(name='John Doe', age=25)
    # Save the user to ElasticSearch
    new_user_id = await user.save()
    print(new_user_id)
```