### Install Development Dependencies

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/contributing.md

Install the core development tools and optional database-specific development dependencies. Use `.[dev]` for general development tools and `.[dev-driver]` for specific database drivers.

```sh
# development tools
pip install .[dev]
# per-database stuff
pip install .[dev-sqlite]
pip install .[dev-postgres]
pip install .[dev-duckdb]
pip install .[dev-mysql]
pip install .[dev-mariadb]
```

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

### Python Example for Bulk Publish

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

This Python code demonstrates how to use the `bulk_publish` method generated by Aiosql for inserting multiple blog entries from a list of dictionaries.

```python
queries = aiosql.from_path("blogs.sql", "psycopg2")
blogs = [
    {"userid": 1, "title": "First Blog", "content": "...", "published": datetime(2018, 1, 1)},
    {"userid": 1, "title": "Next Blog", "content": "...", "published": datetime(2018, 1, 2)},
    {"userid": 2, "title": "Hey, Hey!", "content": "...", "published": datetime(2018, 7, 28)},
]
queries.bulk_publish(conn, blogs)
```

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

### Complete Example: Database Connection and Query Execution

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Demonstrates creating a database connection using `sqlite3`, loading queries from a file using `aiosql.from_path`, and executing a query. The `driver_adapter` argument specifies the database driver.

```pycon
>>> import sqlite3
>>> import aiosql
>>> conn = sqlite3.connect("./blogs.db")
>>> # Note the "sqlite3" driver_adapter argument is what tells
>>> # aiosql it should be expecting a sqlite3 connection object.
>>> queries = aiosql.from_path("./blogs.sql", "sqlite3")
>>> queries.get_all_blogs(conn)
[(1,
  1,
  'What I did Today',
  'I mowed the lawn, washed some clothes, and ate a burger.\n'
  '\n'
  'Until next time,\n'
  'Bob',
  '2017-07-28'),
 (2, 3, 'Testing', 'Is this thing on?\n', '2018-01-01'),
 (3,
  1,
  'How to make a pie.',
  '1. Make crust\n2. Fill\n3. Bake\n4. Eat\n',
  '2018-11-23')]
```

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

### Python Example for Creating Table

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

This Python code shows how to execute a SQL script to create a table using Aiosql. It initializes Aiosql with a SQLite driver and calls the generated `create_table_blogs` method.

```python
queries = aiosql.from_path("create_schema.sql", "sqlite3")
queries.create_table_blogs(conn)
```

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

### Define SQL Queries in a File

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

SQL queries are defined in .sql files using '-- name:' comments to denote method names and ':variable' for substitutions. This example shows two queries: one to get all blogs and another to get user-specific blogs with formatted dates.

```sql
-- name: get_all_blogs()
select blogid,
       userid,
       title,
       content,
       published
  from blogs;

-- name: get_user_blogs(username)
-- Get blogs with a fancy formatted published date and author field
    select b.blogid,
           b.title,
           strftime('%Y-%m-%d %H:%M', b.published) as published,
           u.username as author
      from blogs b
inner join users u on b.userid = u.userid
     where u.username = :username
  order by b.published desc;
```

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

### Python Example for Returning Blog ID

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

This Python code demonstrates calling a query defined with the `$` operator to insert a new blog and retrieve its `blogid`.

```python
blogid = queries.publish_new_blog(conn, userid=1, title="AioSQL New Features", content="…")
```

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

### Run Docker Clients Against Detached Servers

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Starts Docker containers that run AioSQL tests against manually started detached database servers. Ensure the correct host and database are specified.

```shell
docker run -it -v .:/code --add-host=host.docker.internal:host-gateway \
  python-aiosql-dbs \
  make VENV=/venv MA_HOST=host.docker.internal check.pytest.mariadb.detached
```

```shell
docker run -it -v .:/code --add-host=host.docker.internal:host-gateway \
  python-aiosql-mysql \
  make VENV=/venv MY_HOST=host.docker.internal check.pytest.mysql.detached
```

```shell
docker run -it -v .:/code --add-host=host.docker.internal:host-gateway \
  python-aiosql-dbs \
  make VENV=/venv MS_HOST=host.docker.internal check.pytest.mssql.detached
```

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

### Python Example: Using aiosql with SQLite

Source: https://context7.com/nackjicholson/aiosql/llms.txt

This Python code demonstrates how to load SQL queries using aiosql and execute them with a synchronous sqlite3 connection, utilizing various query operators.

```python
import sqlite3
import aiosql
from datetime import date

queries = aiosql.from_path("queries/demo.sql", "sqlite3")

with sqlite3.connect(":memory:") as conn:
    queries.setup_schema(conn)                        # # script operator

    n = queries.insert_blog(conn,
        userid=1, title="Hello World", content="Hi")
    # => 1 (rows affected)

    blogid = queries.insert_implicit_id(conn,
        userid=1, title="Second Post", content="...")
    # => 2 (lastrowid)

    count = queries.scalar_count(conn)                # $ operator
    # => 2

    row = queries.one_row(conn, blogid=1)             # ^ operator
    # => (1, 'Hello World', 'Hi')

    all_rows = list(queries.all_rows(conn))           # default operator
    # => [(1, 'Hello World'), (2, 'Second Post')]

    bulk_data = [
        {"userid": 2, "title": "Bulk A", "content": "..."},
        {"userid": 2, "title": "Bulk B", "content": "..."},
    ]
    queries.bulk_insert(conn, bulk_data)              # *! operator
    conn.commit()
```

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

### Access Raw Database Cursor with `_cursor` Methods

Source: https://context7.com/nackjicholson/aiosql/llms.txt

Use the `<name>_cursor` method to get the raw database cursor. This allows direct access to cursor attributes like `.description` and methods like `.fetchone()` and `.fetchall()`.

```python
import asyncio
import aiosql
import aiosqlite

SQL = """
-- name: get-users()
select user_id, username, name, email from users order by user_id;
"""

queries = aiosql.from_str(SQL, "aiosqlite")

async def inspect_schema():
    async with aiosqlite.connect("app.db") as conn:
        async with queries.get_users_cursor(conn) as cur:
            # Read column metadata from the cursor
            col_names = [col[0] for col in cur.description]
            print("Columns:", col_names)
            # Columns: ['user_id', 'username', 'name', 'email']

            first = await cur.fetchone()
            print("First row:", first)
            # First row: (1, 'alice', 'Alice', 'alice@example.com')

            rest = await cur.fetchall()
            print(f"Remaining {len(rest)} rows")

asyncio.run(inspect_schema())
```

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

### Run MS SQL Server Docker Container

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Starts a detached MS SQL Server container with specified environment variables for EULA acceptance, SA password, and edition. It also maps the default SQL Server port.

```shell
docker run -e "ACCEPT_EULA=Y" -e "MSSQL_SA_PASSWORD=Abc123.." -e "MSSQL_PID=Developer" \
  -p 1433:1433  --name mssqltest --hostname mssqltest -d mcr.microsoft.com/mssql/server:2022-latest
```

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

### Automate Development Tasks with Makefile

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/contributing.md

Utilize the provided Makefile for common development tasks. `make venv.dev` installs the development virtual environment, and `make check` runs all checks including pytest, flake8, and coverage.

```sh
make venv.dev  # install dev virtual environment
source venv/bin/activate
make check  # run all checks: pytest, flake8, coverage…
```

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

### SQL Identifier Substitution

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/todo.md

Example of using HugSQL-like syntax for substituting SQL identifiers, such as column and table names. This requires careful handling of identifier quoting to prevent SQL injection.

```sql
-- name: select(cols, table)
SELECT :i*:cols FROM :i:table ORDER BY 1;
```

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

### Python Example for Implicit Returning Blog ID

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

This Python code shows how to call a query defined with the `<!` operator to insert a blog and retrieve the `blogid` of the newly inserted row, particularly useful for older SQLite versions.

```python
blogid = queries.publish_a_blog(conn, userid=1, title="Hi", content="blah blah.")
```

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

### Insert/Update/Delete with Returning Clause

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

When SQL databases support returning values after modifications (e.g., using `returning`), use operators like `$` (scalar), `^` (tuple), or `!` (relation) to handle the results. This example uses `$` to return a single `blogid`.

```sql
-- name: publish-new-blog(userid, title, content)$
insert into blogs (userid, title, content)
  values (:userid, :title, :content)
  returning blogid;
```

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

### Clone Repository and Set Up Virtual Environment

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/contributing.md

Clone the project repository, navigate into the directory, create a virtual environment, activate it, and upgrade pip. This is the initial step for setting up the development environment.

```sh
# get the project sources
git clone git@github.com:nackjicholson/aiosql.git
cd aiosql
# create a venv manually
python -m venv venv
source venv/bin/activate
pip install --upgrade pip
```

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

### Run a Client Container with Host Access

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Launches an interactive Docker container, mounting the current directory as a volume and adding a host entry for accessing the Docker host from within the container.

```shell
docker run -it -v .:/code --add-host=host.docker.internal:host-gateway some-image bash
```

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

### Use a Registered Database Adapter

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/database-driver-adapters.md

Load SQL queries from a file using a specific registered adapter. Ensure the adapter is registered before attempting to use it.

```python
import aiosql

queries = aiosql.from_path("some.sql", "acmedb")
```

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

### Execute Parameterized Query with Keyword Arguments

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Execute the `get_user_blogs` query using keyword arguments, passing the connection object and the username.

```python
>>> import sqlite3
>>> import aiosql
>>> conn = sqlite3.connect("./blogs.db")
>>> queries = aiosql.from_path("./blogs.sql", "sqlite3")
>>>
>>> # Using keyword args
>>> queries.get_user_blogs(conn, username="bobsmith")
[(3, 'How to make a pie.', '2018-11-23 00:00', 'bobsmith'), (1, 'What I did Today', '2017-07-28 00:00', 'bobsmith')]
```

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

### Load SQL Queries with Namespacing using Subdirectories

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Organize SQL queries into subdirectories to create namespaces. This allows for better organization and avoids naming conflicts when loading queries from multiple files.

```default
example/sql/
        ├── blogs/
        │   └── blogs.sql
        ├── create_schema.sql
        └── users/
            └── users.sql
```

```python
queries = aiosql.from_path("example/sql", "sqlite3")
```

```python
queries.blogs.get_all(conn)
queries.users.get_all(conn)
```

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

### Pull Docker Images and Build Client Images

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Use these commands to pull the necessary official Docker images for databases and Ubuntu, and then build the custom AioSQL client images.

```shell
docker image pull postgres
docker image pull mariadb
docker image pull mysql
docker image pull ubuntu
make docker.aiosql
```

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

### Build a Docker Image

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Builds a Docker image with a specific tag and Dockerfile. This is useful for creating custom application images.

```shell
docker build -t aiosql-python-mysql -f dockerfile.python-mysql .
```

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

### Load SQL Queries from a Directory

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Load all `.sql` files from a specified directory into a single `Queries` object. This is useful for organizing larger projects.

```default
example/sql
        ├── blogs.sql
        ├── create_schema.sql
        └── users.sql
```

```python
queries = aiosql.from_path("example/sql", "sqlite3")
```

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

### View Query Docstring in Python

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

After loading queries, use the `help()` function in Python to view the generated docstring for a query method.

```python
>>> import aiosql
>>> queries = aiosql.from_path("blogs.sql", "sqlite3")
>>> help(queries.get_all_blogs)
Help on method get_all_blogs in module aiosql.queries:

get_all_blogs(conn, *args, **kwargs) method of aiosql.queries.Queries instance
    Fetch all fields for every blog in the database.
```

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

### Load SQL Queries from a File in Python

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Use `aiosql.from_path` to load SQL queries from a specified file or directory. The function takes the path to the SQL file(s) and the database driver name. It returns a `Queries` object with methods corresponding to the SQL query names.

```python
queries = aiosql.from_path("blogs.sql", "sqlite3")
```

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

### Asynchronous Database Interaction with aiosql and aiosqlite

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/index.md

Load SQL queries from a file using aiosql.from_path and execute them asynchronously with aiosqlite. Queries are awaited, and results can be iterated asynchronously.

```python
import asyncio
import aiosql
import aiosqlite

queries = aiosql.from_path("greetings.sql", "aiosqlite")

async def main():
    async with aiosqlite.connect("greetings.db") as conn:
        user = await queries.get_user_by_username(conn, username="willvaughn")

        async for _, greeting in queries.get_all_greetings(conn):
            print(f"{greeting}, {user[2]}!")

asyncio.run(main())
```

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

### Using a Custom Database Driver Adapter

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/database-driver-adapters.md

Pass the custom adapter class to the `driver_adapter` argument when building queries.

```python
queries = aiosql.from_path("foo.sql", driver_adapter=AcmeAdapter)
```

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

### Execute Parameterized Query with Positional Argument

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Execute the `get_user_blogs` query using a positional argument. Be cautious as positional arguments are applied in the order of appearance in the SQL query.

```python
>>> # Using positional argument
>>> queries.get_user_blogs(conn, "janedoe")
[(2, 'Testing', '2018-01-01 00:00', 'janedoe')]
```

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

### SQL Queries for aiosql

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/index.md

Define SQL queries in a .sql file with optional parameter declarations and comments. Use '-- name:' to define query names and '(parameter)' for named parameters.

```sql
-- name: get_all_greetings()
-- Get all the greetings in the database
select greeting_id, greeting
  from greetings
 order by 1;

-- name: get_user_by_username(username)^
-- Get a user from the database using a named parameter
select user_id, username, name
  from users
  where username = :username;
```

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

### Execute Query Using Object Attributes

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Execute the `add_user` query by passing an object `calvin` which has `username` and `name` attributes that correspond to the `:u.username` and `:u.name` placeholders in the SQL.

```python
# User is some class with attributes username and name
calvin = User("calvin", "Calvin")
queries.add_user(u=calvin)
```

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

### Synchronous Database Interaction with aiosql

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/index.md

Load SQL queries from a file using aiosql.from_path and execute them with a synchronous database connection. The connection object is passed to each query call.

```python
import aiosql
import sqlite3

queries = aiosql.from_path("greetings.sql", "sqlite3")

with sqlite3.connect("greetings.db") as conn:
    user = queries.get_user_by_username(conn, username="willvaughn")
    # user: (1, "willvaughn", "William")

    for _, greeting in queries.get_all_greetings(conn):
        # scan: (1, "Hi"), (2, "Aloha"), (3, "Hola"), …
        print(f"{greeting}, {user[2]}!")
    # Hi, William!
    # Aloha, William!
    # …
```

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

### aiosql.from_path

Source: https://context7.com/nackjicholson/aiosql/llms.txt

Loads named SQL queries from a specified file path or directory into a Queries object. Subdirectories are used for namespacing.

```APIDOC
## aiosql.from_path(sql_path, driver_adapter, ...)

### Description
Parses a `.sql` file or an entire directory of `.sql` files and returns a `Queries` object with each named query as a callable method. When a directory is provided, subdirectory names become namespaced attributes on the returned object.

### Parameters
- **sql_path** (string) - Path to the .sql file or directory containing .sql files.
- **driver_adapter** (string) - The name of the database driver adapter to use (e.g., "sqlite3", "psycopg").
- **...

### Request Example
```python
import sqlite3
import aiosql

queries = aiosql.from_path("queries/users.sql", "sqlite3")

with sqlite3.connect("app.db") as conn:
    all_users = list(queries.get_all_users(conn))
    user = queries.get_by_username(conn, username="alice")
    rows_affected = queries.create_user(conn, username="carol", name="Carol")
    conn.commit()
```

### Response
- **Queries object**: An object with named queries as callable methods.
```

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

### Namespaced Queries from Subdirectories with aiosql

Source: https://context7.com/nackjicholson/aiosql/llms.txt

This Python code demonstrates how aiosql organizes queries from SQL files placed in subdirectories, creating nested attributes for namespace separation. It shows how to access and execute these namespaced queries.

```python
import sqlite3
import aiosql

queries = aiosql.from_path("sql/", "sqlite3")

# Inspect what is available
print(queries.available_queries)
# ['blogs.create', 'blogs.delete', 'blogs.get_all', 'blogs.get_by_id',
#  'users.get_all', 'users.get_by_username', 'users.update_name']

with sqlite3.connect("app.db") as conn:
    all_blogs = list(queries.blogs.get_all(conn))
    all_users = list(queries.users.get_all(conn))

    user = queries.users.get_by_username(conn, username="alice")
    blog = queries.blogs.get_by_id(conn, blogid=1)

    queries.blogs.delete(conn, blogid=42)
    conn.commit()
```

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

### Async Queries with aiosql and asyncio

Source: https://context7.com/nackjicholson/aiosql/llms.txt

This Python code shows how to use aiosql with asynchronous drivers like aiosqlite. Query methods become coroutines that must be awaited, and concurrent execution is possible with asyncio.gather.

```python
import asyncio
import aiosql
import aiosqlite

SQL = """
-- name: get-all-greetings()
select greeting_id, greeting from greetings order by 1;

-- name: get-user(username)^
select user_id, username, name from users where username = :username;

-- name: add-greeting(greeting)!
insert into greetings (greeting) values (:greeting);
"""

queries = aiosql.from_path("greetings.sql", "aiosqlite")

async def main():
    async with aiosqlite.connect("greetings.db") as conn:
        # Run two queries concurrently with asyncio.gather
        greetings, user = await asyncio.gather(
            queries.get_all_greetings(conn),
            queries.get_user(conn, username="willvaughn"),
        )
        # greetings: [(1, 'Hi'), (2, 'Aloha'), (3, 'Hola')]
        # user: (1, 'willvaughn', 'William')

        for _, greeting in greetings:
            print(f"{greeting}, {user[2]}!")
        # Hi, William!
        # Aloha, William!

        await queries.add_greeting(conn, greeting="Bonjour")
        await conn.commit()

asyncio.run(main())
```

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

### Validate Required Queries with aiosql

Source: https://context7.com/nackjicholson/aiosql/llms.txt

This snippet demonstrates how to load queries using aiosql and validate that a specific set of required queries are available. It's useful for ensuring critical SQL assets are present at application startup.

```python
print(repr(queries))
# Queries(['count_blogs', 'find_user', 'recent_posts'])

# Useful for validation at startup:
required = {"find_user", "count_blogs", "recent_posts"}
assert required.issubset(set(queries.available_queries)), "Missing required queries!"
```

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

### Python Method Signatures for Loaded Queries

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

When SQL queries are loaded using `aiosql.from_path`, corresponding Python methods are generated. These methods accept parameters defined in the SQL query comments (e.g., `username`) and return a generator yielding query results.

```python
def get_all_blogs(self) -> Generator[Any]:
    pass

def get_user_blogs(self, username: str) -> Generator[Any]:
    pass
```

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

### Run Tests with Pytest

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/contributing.md

Execute all project tests using the pytest framework. Ensure your virtual environment is activated before running this command.

```sh
pytest
```

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

### Inspect Loaded Query Names with `available_queries`

Source: https://context7.com/nackjicholson/aiosql/llms.txt

The `available_queries` property returns a sorted list of all method names loaded into a `Queries` object. This includes namespaced queries from subdirectories.

```python
import aiosql

SQL = """
-- name: find-user(username)^
select * from users where username = :username;

-- name: count-blogs()$
select count(*) from blogs;

-- name: recent-posts(limit)
select title, published from blogs order by published desc limit :limit;
"""

queries = aiosql.from_str(SQL, "sqlite3")

print(queries.available_queries)
```

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

### Access Raw SQL and Operation Type with `.sql` and `.operation`

Source: https://context7.com/nackjicholson/aiosql/llms.txt

The `.sql` and `.operation` attributes provide access to the prepared SQL string and operation type, respectively. This is useful for logging, metrics, tracing, and driver-specific features like `execute_values`.

```python
import time
import logging
import contextlib
import aiosql
import psycopg2
from psycopg2.extras import execute_values

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("db.metrics")

SQL = """
-- name: create-schema#
create table if not exists events (id int primary key, name text, ts timestamptz);

-- name: upsert-events!
INSERT INTO events (id, name, ts) VALUES %s
ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, ts = EXCLUDED.ts;

-- name: get-events
select id, name, ts from events order by ts desc;
"""

queries = aiosql.from_str(SQL, "psycopg2")

@contextlib.contextmanager
def timed_query(fn):
    start = time.perf_counter()
    yield
    elapsed_ms = (time.perf_counter() - start) * 1000
    log.info(f"[{fn.operation.name}] {fn.__name__!r} completed in {elapsed_ms:.2f}ms")
    log.info(f"SQL: {fn.sql!r}")

conn = psycopg2.connect("dbname=myapp user=postgres")
queries.create_schema(conn)

# Use raw .sql with psycopg2's execute_values for bulk upsert
with conn.cursor() as cur:
    execute_values(cur, queries.upsert_events.sql, [
        (1, "user_signup",  "2024-01-15 09:00:00+00"),
        (2, "purchase",     "2024-01-15 10:30:00+00"),
        (3, "user_logout",  "2024-01-15 11:00:00+00"),
    ])

with timed_query(queries.get_events):
    events = list(queries.get_events(conn))
# INFO:db.metrics:[SELECT] 'get_events' completed in 0.84ms
# INFO:db.metrics:SQL: 'select id, name, ts from events order by ts desc;'

print(events)
# [(3, 'user_logout', ...), (2, 'purchase', ...), (1, 'user_signup', ...)]
conn.commit()
conn.close()
```

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

### SQL Query Operators with aiosql

Source: https://context7.com/nackjicholson/aiosql/llms.txt

Demonstrates various aiosql query operators for controlling SQL execution and return types. Operators are appended to the query name in SQL comments.

```sql
-- name: all-rows()
-- Returns a generator of all rows (default, no operator)
select blogid, title from blogs order by blogid;
```

```sql
-- name: one-row(blogid)^
-- Returns the first matching row as a tuple, or None
select blogid, title, content from blogs where blogid = :blogid;
```

```sql
-- name: scalar-count()$
-- Returns the first value of the first row (scalar)
select count(*) from blogs;
```

```sql
-- name: insert-blog(userid, title, content)!
-- Executes insert, returns number of affected rows
insert into blogs (userid, title, content) values (:userid, :title, :content);
```

```sql
-- name: insert-returning(userid, title, content)$
-- Insert with RETURNING (PostgreSQL / modern SQLite)
insert into blogs (userid, title, content)
  values (:userid, :title, :content)
  returning blogid;
```

```sql
-- name: insert-implicit-id(userid, title, content)<!
-- Returns lastrowid (legacy SQLite without RETURNING support)
insert into blogs (userid, title, content) values (:userid, :title, :content);
```

```sql
-- name: bulk-insert(userid, title, content)*!
-- Executes over a sequence of dicts (executemany)
insert into blogs (userid, title, content) values (:userid, :title, :content);
```

```sql
-- name: setup-schema#
-- Executes as a raw SQL script (no parameter substitution)
CREATE TABLE IF NOT EXISTS blogs (
  blogid INTEGER PRIMARY KEY AUTOINCREMENT,
  userid INTEGER NOT NULL,
  title  TEXT NOT NULL,
  content TEXT NOT NULL
);
```

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

### Run Tests with Docker Compose

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Execute AioSQL tests using Docker Compose for either pytest or coverage analysis. Ensure client images are built before running tests.

```shell
make docker.pytest
make docker.coverage
```

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

### Synchronous Database Driver Adapter Protocol

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/database-driver-adapters.md

Defines the interface for synchronous database driver adapters. Implement these methods to integrate a PEP 249 compliant driver with aiosql.

```python
    def process_sql(
        self,
        query_name: str,
        op_type: SQLOperationType,
        sql: str
    ) -> str: ...  # pragma: no cover

    def select(
        self,
        conn: Any,
        query_name: str,
        sql: str,
        parameters: ParamType,
        record_class: Callable|None,
    ) -> Generator[Any, None, None]: ...  # pragma: no cover

    def select_one(
        self,
        conn: Any,
        query_name: str,
        sql: str,
        parameters: ParamType,
        record_class: Callable|None,
    ) -> tuple[Any, ...]|None: ...  # pragma: no cover

    def select_value(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> Any|None: ...  # pragma: no cover

    def select_cursor(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> ContextManager[Any]: ...  # pragma: no cover

    def insert_update_delete(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> int: ...  # pragma: no cover

    def insert_update_delete_many(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> int: ...  # pragma: no cover

    def insert_returning(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> Any|None: ...  # pragma: no cover

    def execute_script(self, conn: Any, sql: str) -> str: ...  # pragma: no cover
```

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

### Register a custom database driver adapter with aiosql

Source: https://context7.com/nackjicholson/aiosql/llms.txt

Use `aiosql.register_adapter` to register a custom adapter class under a string name. This allows referencing it by name in `from_path` or `from_str` calls, and can override built-in adapters.

```python
import aiosql
from aiosql.adapters.generic import GenericAdapter

class OracleAdapter(GenericAdapter):
    """Minimal custom adapter for oracledb (named paramstyle)."""
    def process_sql(self, query_name, op_type, sql):
        # oracledb uses :name syntax natively — no transformation needed
        return sql

# Register globally so any subsequent from_path / from_str call can use "oracle"
aiosql.register_adapter("oracle", OracleAdapter)

import oracledb
conn = oracledb.connect(user="scott", password="tiger", dsn="localhost/xepdb1")

queries = aiosql.from_path("queries/reports.sql", "oracle")
rows = list(queries.monthly_summary(conn, month="2024-01"))
print(rows)
```

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

### Test PEP 249 Driver Compatibility

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/database-driver-adapters.md

Before creating a custom adapter, test if your PEP 249 compliant driver works with aiosql by performing a simple query. Ensure your driver supports 'pyformat' or 'named' paramstyles.

```python
import acmedb  # your PEP 249 driver
import aiosql

conn = acmedb.connect("…")
queries = aiosql.from_str("-- name: add42$
SELECT :n + 42;
", acmedb)
assert queries.add42(conn, n=18) == 60
```

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

### Bulk Insert/Update/Delete Operations

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

The `*!` operator is used to execute a SQL statement over a sequence of items, utilizing the `executemany` method. This is ideal for bulk operations like inserting multiple blog entries at once. The generated methods return the number of affected rows.

```APIDOC
### `*!` Insert/Update/Delete Many

The `*!` operator directs aiosql to execute a SQL statement over all items of a given sequence. Under the hood this calls the `executemany` method of many database drivers.

In aiosql we can use this for a bulk publish method that operates over a list of blog entries.

```sql
-- name: bulk-publish(userid, title, content, published)*!
-- Insert many blogs at once
insert into blogs (userid, title, content, published)
  values (:userid, :title, :content, :published);
```

```python
queries = aiosql.from_path("blogs.sql", "psycopg2")
blogs = [
    {"userid": 1, "title": "First Blog", "content": "...", "published": datetime(2018, 1, 1)},
    {"userid": 1, "title": "Next Blog", "content": "...", "published": datetime(2018, 1, 2)},
    {"userid": 2, "title": "Hey, Hey!", "content": "...", "published": datetime(2018, 7, 28)},
]
queries.bulk_publish(conn, blogs)
```

The methods returns the number of affected rows, if available.
```

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

### Add Documentation to SQL Query

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

Use SQL comments between the name definition and the query code to generate Python docstrings for the query method.

```sql
-- name: get-all-blogs()
-- Fetch all fields for every blog in the database.
select * from blogs;
```

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

### Execute SQL Command in MS SQL Server Container

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Connects to the running MS SQL Server container using sqlcmd to execute SQL commands. Requires the SA username and password.

```shell
docker exec -it mssqltest /opt/mssql-tools18/bin/sqlcmd -C -S localhost -U sa -P "Abc123.."
# type a command
# go
```

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

### aiosql.register_adapter

Source: https://context7.com/nackjicholson/aiosql/llms.txt

Registers a custom database driver adapter class under a given name for use with `from_path` and `from_str`.

```APIDOC
## aiosql.register_adapter(name, adapter)

### Description
Registers a new adapter class under a string name so it can be referenced by name in `from_path` / `from_str` calls. Can also override an existing built-in adapter.

### Parameters
- **name** (string) - The name to register the adapter under (e.g., "oracle").
- **adapter** (class) - The custom adapter class to register.

### Request Example
```python
import aiosql
from aiosql.adapters.generic import GenericAdapter

class OracleAdapter(GenericAdapter):
    def process_sql(self, query_name, op_type, sql):
        return sql

aiosql.register_adapter("oracle", OracleAdapter)

import oracledb
conn = oracledb.connect(user="scott", password="tiger", dsn="localhost/xepdb1")

queries = aiosql.from_path("queries/reports.sql", "oracle")
rows = list(queries.monthly_summary(conn, month="2024-01"))
print(rows)
```

### Response
None. This function modifies the global adapter registry.
```

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

### Load SQL Queries from a String

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Load SQL queries directly from a multi-line string. The `aiosql.from_str` function parses the string and creates a `Queries` object with methods corresponding to the `-- name:` comments in the SQL.

```python
sql_str = """
-- name: get_all_blogs()
select blogid,
       userid,
       title,
       content,
       published
  from blogs;

-- name: get_user_blogs(username)
-- Get blogs with a fancy formatted published date and author field
    select b.blogid,
           b.title,
           strftime('%Y-%m-%d %H:%M', b.published) as published,
           u.username as author
      from blogs b
inner join users u on b.userid = u.userid
     where u.username = :username
  order by b.published desc;
"""

queries = aiosql.from_str(sql_str, "sqlite3")
```

```python
queries.get_all_blogs(conn)
queries.get_user_blogs(conn, username="johndoe")
```

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

### Access Prepared SQL String for Custom Execution

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/advanced-topics.md

Retrieve the raw SQL string of a query using the `.sql` attribute. This allows using aiosql to manage SQL definitions while leveraging database-specific features or libraries like `psycopg2.extras.execute_values` for custom execution.

```python
import aiosql
import psycopg2
from psycopg2.extras import execute_values

SQL = """
-- name: create_schema#
create table if not exists test (id int primary key, v1 int, v2 int);

-- name: insert!
INSERT INTO test (id, v1, v2) VALUES %s;

-- name: update!
UPDATE test SET v1 = data.v1 FROM (VALUES %s) AS data (id, v1)
WHERE test.id = data.id;

-- name: getem
select * from test order by id;
"""

queries = aiosql.from_str(SQL, "psycopg2")
conn = psycopg2.connect("dbname=test")
queries.create_schema(conn)
with conn.cursor() as cur:
    execute_values(cur, queries.insert.sql, [(1, 2, 3), (4, 5, 6), (7, 8, 9)])
    execute_values(cur, queries.update.sql, [(1, 20), (4, 50)])

print(list(queries.getem(conn)))
#  [(1, 20, 3), (4, 50, 6), (7, 8, 9)]
```

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

### Define Parameterized SQL Query

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Define a SQL query with a named parameter `:username`. This query retrieves blog posts, formatting the published date and including the author's username.

```sql
-- name: get_user_blogs(username)
-- Get blogs with a fancy formatted published date and author field
    select b.blogid,
           b.title,
           strftime('%Y-%m-%d %H:%M', b.published) as published,
           u.username as author
      from blogs b
inner join users u on b.userid = u.userid
     where u.username = :username
  order by b.published desc;
```

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

### SQL Query with Named Parameters

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

Define a query that accepts named parameters using `:param_name` syntax. Simple attributes can be referenced with `.`. The generated Python function accepts these as keyword arguments.

```sql
-- name: with-params(name, x)^
select length(:name), :x.real + x.imaj;
```

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

### Define SQL Query for Object Attribute Access

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/getting-started.md

Define an SQL query that inserts a new user, accessing `username` and `name` attributes from an object passed as the `:u` parameter.

```sql
-- name: add_user(u)
insert into users(username, name)
  values (:u.username, :u.name);
```

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

### Bulk Insert/Update/Delete with `*!` Operator

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

The `*!` operator is used for executing SQL statements over a sequence of items, internally calling `executemany`. This is suitable for bulk operations like inserting multiple blog entries.

```sql
-- name: bulk-publish(userid, title, content, published)*!
-- Insert many blogs at once
insert into blogs (userid, title, content, published)
  values (:userid, :title, :content, :published);
```

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

### Python Execution of Substituted SQL

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/todo.md

Python code demonstrating how to execute a query with substituted identifiers. Ensure that the values provided for 'cols' and 'table' are properly sanitized or validated before being passed to the query.

```python
res = db.select(conn, cols=["uid", "name"], table="users")
```

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

### Asynchronous Database Driver Adapter Protocol

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/database-driver-adapters.md

Defines the interface for asynchronous database driver adapters. Implement these methods to integrate an asynchronous driver with aiosql.

```python
    def process_sql(
        self,
        query_name: str,
        op_type: SQLOperationType,
        sql: str
    ) -> str: ...  # pragma: no cover

    def select(
        self,
        conn: Any,
        query_name: str,
        sql: str,
        parameters: ParamType,
        record_class: Callable|None,
    ) -> collections.abc.AsyncGenerator[Any, None]: ...  # pragma: no cover

    async def select_one(
        self,
        conn: Any,
        query_name: str,
        sql: str,
        parameters: ParamType,
        record_class: Callable|None,
    ) -> Any|None: ...  # pragma: no cover

    async def select_value(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> Any|None: ...  # pragma: no cover

    def select_cursor(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> AsyncContextManager[Any]: ...  # pragma: no cover

    # TODO: Next major version introduce a return? Optional return?
    async def insert_update_delete(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> None: ...  # pragma: no cover

    # TODO: Next major version introduce a return? Optional return?
    async def insert_update_delete_many(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> None: ...  # pragma: no cover

    async def insert_returning(
        self, conn: Any, query_name: str, sql: str, parameters: ParamType
    ) -> Any|None: ...  # pragma: no cover

    async def execute_script(self, conn: Any, sql: str) -> str: ...  # pragma: no cover
```

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

### Register a Custom Database Adapter

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/database-driver-adapters.md

Register a custom adapter named 'acmedb' by subclassing `aiosql.Adapter`. This allows aiosql to use your custom database driver.

```python
import aiosql

# Assume AcmeAdapter is defined elsewhere and subclasses aiosql.Adapter
# class AcmeAdapter(aiosql.Adapter):
#     ...

aiosql.register_adapter("acmedb", AcmeAdapter)
```

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

### Pass Objects to Queries Using Dot-Syntax Parameters

Source: https://context7.com/nackjicholson/aiosql/llms.txt

Use dot-syntax (`:obj.attr`) in named parameters to extract attributes directly from Python objects. This simplifies passing complex data structures to queries.

```python
import sqlite3
import aiosql
from dataclasses import dataclass

SQL = """
-- name: create-schema#
CREATE TABLE IF NOT EXISTS users (
  user_id INTEGER PRIMARY KEY AUTOINCREMENT,
  username TEXT UNIQUE NOT NULL,
  email    TEXT NOT NULL,
  role     TEXT NOT NULL DEFAULT 'viewer'
);

-- name: upsert-user(u)!
INSERT INTO users (username, email, role)
  VALUES (:u.username, :u.email, :u.role)
  ON CONFLICT (username) DO UPDATE
    SET email = excluded.email, role = excluded.role;

-- name: get-user(username)^
SELECT user_id, username, email, role FROM users WHERE username = :username;
"""

@dataclass
class User:
    username: str
    email: str
    role: str = "viewer"

queries = aiosql.from_str(SQL, "sqlite3", kwargs_only=False)

with sqlite3.connect(":memory:") as conn:
    queries.create_schema(conn)

    alice = User("alice", "alice@example.com", "admin")
    bob   = User("bob",   "bob@example.com")

    queries.upsert_user(conn, u=alice)
    queries.upsert_user(conn, u=bob)
    conn.commit()

    row = queries.get_user(conn, username="alice")
    print(row)
    # (1, 'alice', 'alice@example.com', 'admin')
```

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

### Execute SQL Scripts

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

The `#` operator is used to execute SQL statements as a script, typically for data definition statements like `CREATE TABLE`. Variable substitution is not supported with this operator.

```APIDOC
### `#` Execute Scripts

Using this operarator will execute sql statements as a script. You can’t do variable substitution with the `#` operator. An example usecase is using data definition statements like create table in order to setup a database.

```sql
-- name: create-table-blogs#
CREATE TABLE IF NOT EXISTS blogs(
  blogid SERIAL PRIMARY KEY,
  userid INTEGER NOT NULL REFERENCES users,
  title TEXT NOT NULL,
  content TEXT NOT NULL,
  published DATE NOT NULL DEFAULT CURRENT_DATE
);
```

```python
queries = aiosql.from_path("create_schema.sql", "sqlite3")
queries.create_table_blogs(conn)
```

Note: SQL scripts do not accept parameters.
```

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

### Insert/Update/Delete with Implicit Returning for SQLite

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

The `<!` operator is for insert/update/delete operations that implicitly return a value, such as the `blogid` of an inserted row in older SQLite versions via `cur.lastrowid`. For modern SQLite supporting `returning`, use explicit clauses with other operators.

```sql
-- name: publish-a-blog(userid, title, content)<!
insert into blogs(userid, title, content)
  values (:userid, :title, :content);
```

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

### Call SQL Query with Named Parameters

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

Call the generated Python function with named parameters corresponding to the SQL query's parameters.

```python
res = queries.with_params(name="Calvin", x=(1+1j))
# => (6, 2.0)
```

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

### Execute SQL Scripts with `#` Operator

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

Use the `#` operator to execute SQL statements as a script, typically for data definition language (DDL) statements like `CREATE TABLE`. This operator does not support variable substitution.

```sql
-- name: create-table-blogs#
CREATE TABLE IF NOT EXISTS blogs(
  blogid SERIAL PRIMARY KEY,
  userid INTEGER NOT NULL REFERENCES users,
  title TEXT NOT NULL,
  content TEXT NOT NULL,
  published DATE NOT NULL DEFAULT CURRENT_DATE
);
```

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

### Pull MS SQL Server Docker Image

Source: https://github.com/nackjicholson/aiosql/blob/main/docker/README.md

Pulls the latest version of the MS SQL Server Docker image from Microsoft's container registry.

```shell
docker pull mcr.microsoft.com/mssql/server:2022-latest
```

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

### Declare SQL Query Parameters

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

Declare query parameter names in parentheses after the method name to enable validation against unused or undeclared parameters.

```sql
-- name: search(title, published)
select title from blogs where title LIKE :title and published = :published;
```

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

### Observe SQL Operation Type and Execution Time

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/advanced-topics.md

Access the SQL operation type (e.g., 'SELECT') and the raw SQL string via the query function's attributes. This enables custom observability logic, such as logging operation details and execution times, for metrics, tracing, or logging purposes.

```python
import time
import logging
import contextlib
import pg_execute_values as pev

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("metrics")

def report_metrics(op, sql, op_time):
    log.info(f"Operation: {op.name!r}\nSQL: {sql!r} 
Time (ms): {op_time}")

@contextlib.contextmanager
def observe_query(func):
    op = func.operation
    sql = func.sql
    start = time.time()
    yield
    end = time.time()
    op_time = end - start
    report_metrics(op, sql, op_time)

with observe_query(pev.queries.getem):
    pev.queries.getem(pev.conn)

# INFO:metrics:Operation: 'SELECT'
# SQL: 'select * from test order by id;' 
# Time (ms): 2.6226043701171875e-06
```

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

### Define SQL Query Name

Source: https://github.com/nackjicholson/aiosql/blob/main/docs/source/defining-sql-queries.md

Define a query name using a SQL comment `-- name: query-name()`. Dashes in the name are converted to underscores in the Python method name.

```sql
-- name: get-all-blogs()
select * from blogs;
```