### AIOMySQL Checkpoint Saver Setup (Asynchronous)

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Illustrates the asynchronous setup of the AIOMySQLSaver for non-blocking checkpointing in LangGraph. This is suitable for high-throughput applications requiring async I/O.

```python
from langgraph.checkpoint.mysql.aio import AIOMySQLSaver
async with AIOMySQLSaver.from_conn_string(tcp_uri) as saver:
    await saver.setup()
```

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

### PyMySQL Checkpoint Saver Setup (Synchronous)

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Demonstrates how to set up and use the PyMySQLSaver for synchronous checkpointing in LangGraph. It utilizes a connection string and the context manager pattern for resource management.

```python
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver
with PyMySQLSaver.from_conn_string(tcp_uri) as saver:
    saver.setup()
```

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

### AsyncMySaver: Async MySQL Checkpoint Saver Setup and Usage

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Demonstrates how to initialize and use the AsyncMySaver for asynchronous checkpoint operations with MySQL. It shows setting up the saver, putting a checkpoint, and retrieving a specific checkpoint by its ID using the asyncmy driver.

```python
import asyncio
from langgraph.checkpoint.mysql.asyncmy import AsyncMySaver

DB_URI = "mysql://user:password@localhost:3306/mydb"

async def main():
    async with AsyncMySaver.from_conn_string(DB_URI) as checkpointer:
        await checkpointer.setup()

        config = {"configurable": {"thread_id": "asyncmy-thread", "checkpoint_ns": ""}}

        checkpoint = {
            "v": 4,
            "ts": "2024-08-01T10:00:00.000000+00:00",
            "id": "asyncmy-cp-001",
            "channel_values": {"state": "processing"},
            "channel_versions": {"state": 1},
            "versions_seen": {},
        }

        await checkpointer.aput(config, checkpoint, {"step": 1}, {})

        # Retrieve specific checkpoint by ID
        specific_config = {
            "configurable": {
                "thread_id": "asyncmy-thread",
                "checkpoint_ns": "",
                "checkpoint_id": "asyncmy-cp-001"
            }
        }
        result = await checkpointer.aget_tuple(specific_config)
        print(result.checkpoint)

asyncio.run(main())
```

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

### PyMySQLSaver: List and Filter Checkpoints

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Shows how to query and list checkpoints using PyMySQLSaver, with examples of filtering by metadata, limiting results, and retrieving checkpoints before a specific point in time. This enables efficient retrieval of historical checkpoint data.

```python
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver

DB_URI = "mysql://user:password@localhost:3306/mydb"

with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    config = {"configurable": {"thread_id": "user-123"}}

    # List all checkpoints for a thread (newest first)
    all_checkpoints = list(checkpointer.list(config))

    # List with limit
    recent = list(checkpointer.list(config, limit=5))

    # Filter by metadata
    filtered = list(checkpointer.list(
        config,
        filter={"source": "input", "step": 1}
    ))

    # Get checkpoints before a specific checkpoint
    before_config = {
        "configurable": {
            "thread_id": "user-123",
            "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"
        }
    }
    older = list(checkpointer.list(config, before=before_config, limit=10))

    for cp in recent:
        print(f"ID: {cp.config['configurable']['checkpoint_id']}")
        print(f"Metadata: {cp.metadata}")
        print(f"Parent: {cp.parent_config}")
```

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

### AIOMySQLStore: Asynchronous Key-Value Store Operations

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Illustrates asynchronous key-value storage operations using AIOMySQLStore for non-blocking data handling in async Python applications. It supports asynchronous batch operations for put, get, and search, requiring a MySQL connection string and table setup.

```python
import asyncio
from langgraph.store.mysql.aio import AIOMySQLStore
from langgraph.store.base import PutOp, GetOp, SearchOp

DB_URI = "mysql://user:password@localhost:3306/mydb"

async def main():
    async with AIOMySQLStore.from_conn_string(DB_URI) as store:
        await store.setup()

        # Async batch operations
        await store.abatch([
            PutOp(namespace=("sessions", "abc123"), key="state", value={"step": 1, "data": [1, 2, 3]}),
            PutOp(namespace=("sessions", "abc123"), key="context", value={"user_id": "u-001"}),
        ])

        # Async get
        results = await store.abatch([
            GetOp(namespace=("sessions", "abc123"), key="state"),
        ])
        print(results[0].value)  # {'step': 1, 'data': [1, 2, 3]}

        # Async search with comparison operators
        search_results = await store.abatch([
            SearchOp(
                namespace_prefix=("sessions",),
                filter={"step": {"$gte": 1}},
                limit=20,
                offset=0
            )
        ])
        for item in search_results[0]:
            print(f"{item.key}: {item.value}")

asyncio.run(main())
```

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

### Asynchronous Checkpoint Saver with AIOMySQL

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Provides an asynchronous checkpoint saver using the aiomysql driver for non-blocking checkpoint operations in async applications. It supports async setup, aput, aget_tuple, alist, and aput_writes.

```python
import asyncio
from langgraph.checkpoint.mysql.aio import AIOMySQLSaver

DB_URI = "mysql://user:password@localhost:3306/mydb"

async def main():
    async with AIOMySQLSaver.from_conn_string(DB_URI) as checkpointer:
        # Initialize database tables
        await checkpointer.setup()

        config = {"configurable": {"thread_id": "async-thread-1", "checkpoint_ns": ""}}

        checkpoint = {
            "v": 4,
            "ts": "2024-07-31T20:14:19.804150+00:00",
            "id": "async-checkpoint-001",
            "channel_values": {
                "user_input": "What is the weather?",
                "agent_response": "Let me check that for you."
            },
            "channel_versions": {"user_input": 1, "agent_response": 2},
            "versions_seen": {},
        }

        # Async put operation
        saved_config = await checkpointer.aput(
            config, checkpoint,
            {"source": "agent", "step": 2},
            {}
        )

        # Async get operation
        result = await checkpointer.aget_tuple(config)
        print(result.checkpoint["channel_values"])
        # Output: {'user_input': 'What is the weather?', 'agent_response': 'Let me check that for you.'}

        # Async list with filtering
        async for cp in checkpointer.alist(config, limit=10):
            print(f"Found checkpoint: {cp.config['configurable']['checkpoint_id']}")

        # Store intermediate writes for a task
        await checkpointer.aput_writes(
            saved_config,
            writes=[("tool_result", {"weather": "sunny", "temp": 72})],
            task_id="weather-tool-123"
        )

asyncio.run(main())

```

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

### GET /checkpoints/list

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Retrieves a list of checkpoints for a thread with support for filtering, pagination, and time-based constraints.

```APIDOC
## GET /checkpoints/list

### Description
Queries checkpoints for a given thread. Supports limiting results, filtering by metadata, and retrieving checkpoints before a specific ID.

### Method
GET

### Endpoint
PyMySQLSaver.list(config, limit, filter, before)

### Parameters
#### Query Parameters
- **config** (dict) - Required - Thread configuration.
- **limit** (int) - Optional - Maximum number of checkpoints to return.
- **filter** (dict) - Optional - Metadata key-value pairs to filter by.
- **before** (dict) - Optional - Configuration object to retrieve checkpoints prior to a specific checkpoint_id.

### Response
#### Success Response (200)
- **checkpoints** (list) - A list of checkpoint tuples containing config, checkpoint, and metadata.

### Response Example
[
  {
    "config": {"configurable": {"thread_id": "user-123", "checkpoint_id": "1ef4f797..."}},
    "metadata": {"source": "input", "step": 1}
  }
]
```

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

### PyMySQLStore: Synchronous Key-Value Store Operations

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Demonstrates synchronous key-value storage operations using PyMySQLStore. This class provides namespace support for organizing data in MySQL and supports batch operations for put, get, search, and listing namespaces. It requires a MySQL connection string and initializes necessary tables.

```python
from langgraph.store.mysql.pymysql import PyMySQLStore

DB_URI = "mysql://user:password@localhost:3306/mydb"

with PyMySQLStore.from_conn_string(DB_URI) as store:
    # Initialize store tables
    store.setup()

    # Store data with namespaces
    from langgraph.store.base import PutOp, GetOp, SearchOp, ListNamespacesOp

    # Put operations - store user preferences
    store.batch([
        PutOp(namespace=("users", "user-123"), key="preferences", value={"theme": "dark", "language": "en"}),
        PutOp(namespace=("users", "user-123"), key="profile", value={"name": "John", "email": "john@example.com"}),
        PutOp(namespace=("users", "user-456"), key="preferences", value={"theme": "light", "language": "es"}),
    ])

    # Get operations - retrieve specific items
    results = store.batch([
        GetOp(namespace=("users", "user-123"), key="preferences"),
        GetOp(namespace=("users", "user-123"), key="profile"),
    ])
    print(results[0].value)  # {'theme': 'dark', 'language': 'en'}
    print(results[1].value)  # {'name': 'John', 'email': 'john@example.com'}

    # Search with filters
    search_results = store.batch([
        SearchOp(
            namespace_prefix=("users",),
            filter={"theme": "dark"},
            limit=10
        )
    ])
    for item in search_results[0]:
        print(f"Found: {item.namespace}/{item.key} = {item.value}")

    # List namespaces
    namespaces = store.batch([
        ListNamespacesOp(match_conditions=None, max_depth=2, limit=100, offset=0)
    ])
    print(namespaces[0])  # [('users', 'user-123'), ('users', 'user-456')]

    # Delete by setting value to None
    store.batch([
        PutOp(namespace=("users", "user-123"), key="preferences", value=None)
    ])
```

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

### MySQL Connection String Formats

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Provides examples of supported MySQL connection string formats for establishing connections via TCP and Unix sockets. These URIs are used to configure database access for LangGraph's MySQL storage and checkpointing components.

```python
# Standard TCP connection
tcp_uri = "mysql://username:password@hostname:3306/database_name"

# Unix socket connection
socket_uri = "mysql://username:password@localhost/database_name?unix_socket=/var/run/mysqld/mysqld.sock"
```

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

### Synchronous Checkpoint Saver with PyMySQL

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Implements a synchronous checkpoint saver using the PyMySQL driver for storing and retrieving LangGraph checkpoints in MySQL. It requires MySQL >= 8.0.19 or MariaDB >= 10.7.1 and supports setup, put, get_tuple, and list operations.

```python
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver

DB_URI = "mysql://user:password@localhost:3306/mydb"

# Create checkpointer from connection string
with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    # Initialize database tables on first use
    checkpointer.setup()

    # Configuration for checkpoint operations
    write_config = {"configurable": {"thread_id": "conversation-1", "checkpoint_ns": ""}}
    read_config = {"configurable": {"thread_id": "conversation-1"}}

    # Store a checkpoint
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "messages": ["Hello", "How can I help?"],
            "current_node": "assistant"
        },
        "channel_versions": {
            "__start__": 2,
            "messages": 3,
            "current_node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {"__start__": 1}
        },
    }
    metadata = {"source": "input", "step": 1, "writes": {"messages": "Hello"}}

    saved_config = checkpointer.put(write_config, checkpoint, metadata, {})
    # Returns: {'configurable': {'thread_id': 'conversation-1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}

    # Retrieve the latest checkpoint
    checkpoint_tuple = checkpointer.get_tuple(read_config)
    print(checkpoint_tuple.checkpoint["channel_values"])
    # Output: {'messages': ['Hello', 'How can I help?'], 'current_node': 'assistant'}

    # List checkpoints with pagination
    checkpoints = list(checkpointer.list(read_config, limit=5))
    for cp in checkpoints:
        print(f"Checkpoint: {cp.config['configurable']['checkpoint_id']}")

```

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

### Synchronous MySQL Checkpoint Saving and Loading with PyMySQLSaver

Source: https://github.com/tjni/langgraph-checkpoint-mysql/blob/main/README.md

Demonstrates how to use PyMySQLSaver for synchronous checkpoint operations. It covers initializing the saver, setting up the database tables, storing a checkpoint, retrieving a checkpoint, and listing available checkpoints. Requires `langgraph-checkpoint-mysql[pymysql]` and a MySQL database connection string.

```python
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

DB_URI = "mysql://mysql:mysql@localhost:3306/mysql"
with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    # call .setup() the first time you're using the checkpointer
    checkpointer.setup()
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
            "__start__": 1
            },
            "node": {
            "start:node": 2
            }
        },
    }

    # store checkpoint
    checkpointer.put(write_config, checkpoint, {}, {})

    # load checkpoint
    checkpointer.get(read_config)

    # list checkpoints
    list(checkpointer.list(read_config))
```

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

### Asynchronous MySQL Checkpoint Saving and Loading with AIOMySQLSaver

Source: https://github.com/tjni/langgraph-checkpoint-mysql/blob/main/README.md

Demonstrates how to use AIOMySQLSaver for asynchronous checkpoint operations. It covers initializing the saver, storing a checkpoint, retrieving a checkpoint, and listing available checkpoints asynchronously. Requires `langgraph-checkpoint-mysql[aiomysql]` and a MySQL database connection string.

```python
from langgraph.checkpoint.mysql.aio import AIOMySQLSaver

async with AIOMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
            "__start__": 1
            },
            "node": {
            "start:node": 2
            }
        },
    }

    # store checkpoint
    await checkpointer.aput(write_config, checkpoint, {}, {})

    # load checkpoint
    await checkpointer.aget(read_config)

    # list checkpoints
    [c async for c in checkpointer.alist(read_config)]
```

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

### Direct PyMySQL Connection for Checkpoint Saver

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Explains how to use a pre-existing PyMySQL connection object to instantiate and set up the PyMySQLSaver. This allows for more control over the database connection lifecycle.

```python
import pymysql
conn = pymysql.connect(
    host="localhost",
    user="username",
    password="password",
    database="database_name",
    autocommit=True  # Required for setup() to work correctly
)
saver = PyMySQLSaver(conn)
saver.setup()
```

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

### LangGraph Integration: PyMySQLSaver for Persistent State

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Shows how to integrate PyMySQLSaver with LangGraph to enable persistent state management for graph executions. This involves setting up the saver, compiling the graph with the checkpointer, and invoking the graph with a thread ID for state persistence and retrieval.

```python
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

# Define state schema
class AgentState(TypedDict):
    messages: Annotated[list[str], operator.add]
    current_step: str

# Build a simple graph
def process_input(state: AgentState) -> dict:
    return {"messages": ["Processed input"], "current_step": "complete"}

builder = StateGraph(AgentState)
builder.add_node("process", process_input)
builder.add_edge(START, "process")
builder.add_edge("process", END)

DB_URI = "mysql://user:password@localhost:3306/mydb"

with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    # Compile graph with checkpointer
    graph = builder.compile(checkpointer=checkpointer)

    # Run graph with thread ID for persistence
    config = {"configurable": {"thread_id": "my-conversation"}}

    result = graph.invoke(
        {"messages": ["Hello"], "current_step": "start"},
        config
    )
    print(result)
    # Output: {'messages': ['Hello', 'Processed input'], 'current_step': 'complete'}

    # Resume from checkpoint in a new session
    state = graph.get_state(config)
    print(f"Persisted state: {state.values}")
```

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

### POST /checkpoints/aput

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Saves a new checkpoint for a specific thread using the asynchronous MySQL driver.

```APIDOC
## POST /checkpoints/aput

### Description
Saves a checkpoint state to the MySQL database asynchronously. This method is used to persist the current state of a graph execution.

### Method
POST

### Endpoint
AsyncMySaver.aput(config, checkpoint, metadata, new_versions)

### Parameters
#### Request Body
- **config** (dict) - Required - Configuration containing thread_id and checkpoint_ns.
- **checkpoint** (dict) - Required - The checkpoint state object.
- **metadata** (dict) - Required - Metadata associated with the checkpoint.
- **new_versions** (dict) - Required - Channel versions seen.

### Request Example
{
  "config": {"configurable": {"thread_id": "asyncmy-thread", "checkpoint_ns": ""}},
  "checkpoint": {"v": 4, "id": "asyncmy-cp-001", "channel_values": {"state": "processing"}},
  "metadata": {"step": 1},
  "new_versions": {}
}
```

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

### PyMySQLSaver: Delete Thread Checkpoints

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Illustrates how to remove all checkpoints, blobs, and associated writes for a specific thread using the PyMySQLSaver. This is useful for cleaning up data related to a particular execution thread.

```python
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver

DB_URI = "mysql://user:password@localhost:3306/mydb"

with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    # Create some checkpoints for a thread
    config = {"configurable": {"thread_id": "temp-thread", "checkpoint_ns": ""}}
    checkpoint = {
        "v": 4, "ts": "2024-08-01T10:00:00+00:00", "id": "temp-cp-1",
        "channel_values": {}, "channel_versions": {}, "versions_seen": {}
    }
    checkpointer.put(config, checkpoint, {}, {})

    # Verify checkpoint exists
    assert checkpointer.get_tuple(config) is not None

    # Delete all data for the thread
    checkpointer.delete_thread("temp-thread")

    # Verify deletion
    assert checkpointer.get_tuple(config) is None
    print("Thread successfully deleted")
```

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

### DELETE /threads/{thread_id}

Source: https://context7.com/tjni/langgraph-checkpoint-mysql/llms.txt

Removes all checkpoints, blobs, and writes associated with a specific thread ID.

```APIDOC
## DELETE /threads/{thread_id}

### Description
Deletes all data associated with a specific thread, effectively clearing the history for that execution path.

### Method
DELETE

### Endpoint
PyMySQLSaver.delete_thread(thread_id)

### Parameters
#### Path Parameters
- **thread_id** (string) - Required - The unique identifier for the thread to be deleted.

### Request Example
{
  "thread_id": "temp-thread"
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.