Skip to content

sqly.sql

ASQL (SQL)

Source code in sqly/sql.py 
class ASQL(SQL):
    async def execute(
        self, connection: Any, query: str | Iterator, data: Optional[Mapping] = None
    ):
        try:
            cursor = await connection.execute(*self.render(query, data))
        except Exception as exc:
            # If the connection is a cursor, get the underlying connection to rollback,
            # because cursors don't have a rollback method.
            if hasattr(connection, "rollback"):
                try:
                    await connection.rollback()
                except Exception:
                    ...
            raise exc

        return cursor

    async def select(
        self,
        connection: Any,
        query: str | Iterator,
        data: Optional[Mapping] = None,
        Constructor=dict,
    ):
        cursor = await self.execute(connection, query, data)
        fields = [d[0] for d in cursor.description]
        async for row in cursor:
            yield Constructor(**dict(zip(fields, row)))

    async def select_one(
        self,
        connection: Any,
        query: str | Iterator,
        data: Optional[Mapping] = None,
        Constructor=dict,
    ):
        records = self.select(
            connection=connection, query=query, data=data, Constructor=Constructor
        )
        async for record in records:
            return record

    async def select_all(
        self,
        connection: Any,
        query: str | Iterator,
        data: Optional[Mapping] = None,
        Constructor=dict,
    ):
        records = []
        async for record in self.select(
            connection=connection, query=query, data=data, Constructor=Constructor
        ):
            records.append(record)

        return records

execute(self, connection, query, data=None) async

Execute the given query on the connection and return the connection cursor.

If the query fails: Rollback the connection and re-raise the exception, as a convenience to the user not to leave the connection in an unusable state.

If .execute() is called with a previously-generated cursor, that cursor will be reused and the same cursor returned from the method call.

Parameters:

Name Type Description Default
connection Connection | Cursor

A DB-API 2.0 compliant database connection or cursor.

 required
query str | Iterator

A query that will be rendered with the given data.

 required
data Optional[Mapping]

A data mapping that will be rendered as params with the query. Optional, but required if the query contains parameters.

 None

Returns:

Type Description
cursor (Cursor)

A DB-API 2.0 compliant database cursor.
Source code in sqly/sql.py 
async def execute(
    self, connection: Any, query: str | Iterator, data: Optional[Mapping] = None
):
    try:
        cursor = await connection.execute(*self.render(query, data))
    except Exception as exc:
        # If the connection is a cursor, get the underlying connection to rollback,
        # because cursors don't have a rollback method.
        if hasattr(connection, "rollback"):
            try:
                await connection.rollback()
            except Exception:
                ...
        raise exc

    return cursor

select(self, connection, query, data=None, Constructor=<class 'dict'>)

Execute the given query on the connection, and yield result records.

The .select() method is a generator which iterates over a native database cursor. The results of the method can be cast to a list via list(...) or can be iterated through one at a time.

If the query fails: Rollback the connection and re-raise the exception, as a convenience to the user not to leave the connection in an unusable state.

Parameters:

Name Type Description Default
connection Connection | Cursor

A DB-API 2.0 compliant database connection or cursor.

 required
query str | Iterator

A query that will be rendered with the given data.

 required
data Optional[Mapping]

A data mapping that will be rendered as params with the query. Optional, but required if the query contains parameters.

 None
Constructor class

A constructor to use to build records from the results. The constructor must take the results of zip(keys, values) as its argument.

 <class 'dict'>

Yields:

Type Description
record (Mapping)

A mapping object that contains a database record.
Source code in sqly/sql.py 
async def select(
    self,
    connection: Any,
    query: str | Iterator,
    data: Optional[Mapping] = None,
    Constructor=dict,
):
    cursor = await self.execute(connection, query, data)
    fields = [d[0] for d in cursor.description]
    async for row in cursor:
        yield Constructor(**dict(zip(fields, row)))

SQL dataclass

Render and execute SQL queries with a given database dialect. All queries are rendered according to the requirements of that dialect.

  • Create queries using the ergonomic “named” parameter format (:key).
  • render() the query to the parameter format native to the current database Dialect.
  • Use the execute() and select() query methods to render() and execute queries with “named” parameters on the given database connection.

Parameters:

Name Type Description Default
dialect Dialect

The SQL dialect used by this database.

 required

Methods:

  • render(): Render a query and accompanying data to database Dialect-native form.
  • execute(): Execute a query on the given connection.
  • select(): Execute a query and select the results as record objects.

Examples:

Initialize the SQL and Connection instances:

>>> import sqlite3
>>> from sqly import SQL
>>> connection = sqlite3.connect(":memory:")
>>> sql = SQL(dialect="sqlite")

Create a table to make queries with:

>>> cursor = sql.execute(connection,
...     "CREATE TABLE widgets (id int, sku varchar)")

Insert a widget:

>>> widget = {"id": 1, "sku": "COG-01"}
>>> cursor = sql.execute(cursor,  # <-- cursor can be re-used
...     "INSERT INTO widgets VALUES (:id, :sku)", widget)
>>> connection.commit()

Select matching widgets:

>>> records = sql.select(connection,
...     "SELECT * FROM widgets WHERE sku like :sku", {"sku": "COG-%"})
>>> for record in records: print(record)
{'id': 1, 'sku': 'COG-01'}
Source code in sqly/sql.py 
@dataclass
class SQL:
    """
    Render and execute SQL queries with a given database dialect. All queries are
    rendered according to the requirements of that dialect.

    * Create queries using the ergonomic "named" parameter format (:key).
    * `render()` the query to the parameter format native to the current database
      Dialect.
    * Use the `execute()` and `select()` query methods to `render()` and execute queries
      with "named" parameters on the given database connection.

    Arguments:
        dialect (Dialect): The SQL [dialect](sqly.dialect.md) used by this database.

    **Methods:**

    * [render()](./#sqly.sql.SQL.render): Render a query and accompanying data to
      database Dialect-native form.
    * [execute()](./#sqly.sql.SQL.execute): Execute a query on the given connection.
    * [select()](./#sqly.sql.SQL.select): Execute a query and select the results as
        record objects.

    Examples:
        Initialize the SQL and Connection instances:

        >>> import sqlite3
        >>> from sqly import SQL
        >>> connection = sqlite3.connect(":memory:")
        >>> sql = SQL(dialect="sqlite")

        Create a table to make queries with:

        >>> cursor = sql.execute(connection,
        ...     "CREATE TABLE widgets (id int, sku varchar)")

        Insert a widget:

        >>> widget = {"id": 1, "sku": "COG-01"}
        >>> cursor = sql.execute(cursor,  # <-- cursor can be re-used
        ...     "INSERT INTO widgets VALUES (:id, :sku)", widget)
        >>> connection.commit()

        Select matching widgets:

        >>> records = sql.select(connection,
        ...     "SELECT * FROM widgets WHERE sku like :sku", {"sku": "COG-%"})
        >>> for record in records: print(record)
        {'id': 1, 'sku': 'COG-01'}

    """

    dialect: Dialect
    queries = queries
    Q = Q

    def __post_init__(self):
        if not isinstance(self.dialect, Dialect):
            self.dialect = Dialect(self.dialect)

    def render(self, query, data=None):
        """
        Render a query string and its parameters for this SQL dialect.

        Arguments:
            query (str | Iterator): a string or iterator of strings.
            data (Mapping): a keyword dict used to render the query parameters.

        Returns:
            (str): the rendered query string.
            (tuple | dict): depends on the param format:

                - positional param formats (QMARK, NUMBERED) return a tuple of values
                - named param formats (NAMED, PYFORMAT) return a dict
        """
        # ordered list of fields for positional outputs (closure for replace_parameter)
        fields = []

        def replace_parameter(match):
            field = match.group(1)

            # Build the ordered fields list
            if self.dialect.param_format.is_positional or field not in fields:
                fields.append(field)

            # Return the field formatted for the param format type
            if self.dialect.param_format == ParamFormat.NAMED:
                return f":{field}"
            elif self.dialect.param_format == ParamFormat.PYFORMAT:
                return f"%({field})s"
            elif self.dialect.param_format == ParamFormat.QMARK:
                return "?"
            elif self.dialect.param_format == ParamFormat.NUMBERED:
                return f"${len(fields)}"
            else:  # self.dialect.param_format == ParamFormat.FORMAT:
                return "%s"

        # 1. Convert query to a string
        if isinstance(query, str):
            query_str = str(query)
        elif hasattr(query, "__iter__"):
            query_str = "\n".join(str(q) for q in walk(query))
        else:
            raise ValueError(f"Query has unsupported type: {type(query)}")

        # 2. Escape string parameters in the PYFORMAT param format
        if self.dialect.param_format == ParamFormat.PYFORMAT:
            # any % must be intended as literal and must be doubled
            query_str = query_str.replace("%", "%%")

        # 3. Replace the parameter with its dialect-specific representation
        pattern = r"(?<!\\):(\w+)\b"  # colon + word not preceded by a backslash
        query_str = re.sub(pattern, replace_parameter, query_str).strip()

        # 4. Un-escape remaining escaped colon params
        if self.dialect.param_format == ParamFormat.NAMED:
            # replace \:word with :word because the colon-escape is no longer needed.
            query_str = re.sub(r"\\:(\w+)\b", r":\1", query_str)

        # 5. Build the parameter_values dict or list for use with the query
        if self.dialect.param_format.is_positional:
            # parameter_values is a list of values
            parameter_values = [
                json.dumps(val) if isinstance(val, dict) else val
                for val in [data[field] for field in fields]
            ]
        else:
            # parameter_values is a dict of key:value fields
            parameter_values = {
                # (dict, set, tuple) for json/b, but list is for "IN / ANY()" params.
                key: json.dumps(val) if isinstance(val, (dict, set, tuple)) else val
                for key, val in {field: data[field] for field in fields}.items()
            }

        # 6. Return a tuple formatted for this Dialect
        if self.dialect == Dialect.ASYNCPG:
            # asyncpg expects the parameters in a tuple following the query string.
            return tuple([query_str] + parameter_values)
        else:
            # other dialects expect the parameters in the second tuple item.
            return (query_str, parameter_values)

    def execute(
        self, connection: Any, query: str | Iterator, data: Optional[Mapping] = None
    ):
        """
        Execute the given query on the connection and return the connection cursor.

        If the query fails: Rollback the connection and re-raise the exception, as a
        convenience to the user not to leave the connection in an unusable state.

        If `.execute()` is called with a previously-generated cursor, that cursor will
        be reused and the same cursor returned from the method call.

        Parameters:
            connection (Connection | Cursor): A DB-API 2.0 compliant database connection
                or cursor.
            query (str | Iterator): A query that will be rendered with the given data.
            data (Optional[Mapping]): A data mapping that will be rendered as params
                with the query. Optional, but required if the query contains parameters.

        Returns:
            cursor (Cursor): A DB-API 2.0 compliant database cursor.
        """
        try:
            cursor = connection.execute(*self.render(query, data))
        except Exception as exc:
            # If the connection is a cursor, get the underlying connection to rollback,
            # because cursors don't have a rollback method.
            if hasattr(connection, "rollback"):
                try:
                    connection.rollback()
                except Exception:
                    ...
            raise exc

        return cursor

    def select(
        self,
        connection: Any,
        query: str | Iterator,
        data: Optional[Mapping] = None,
        Constructor=dict,
    ):
        """
        Execute the given query on the connection, and yield result records.

        The `.select()` method is a generator which iterates over a native database
        cursor. The results of the method can be cast to a list via `list(...)` or can
        be iterated through one at a time.

        If the query fails: Rollback the connection and re-raise the exception, as a
        convenience to the user not to leave the connection in an unusable state.

        Parameters:
            connection (Connection | Cursor): A DB-API 2.0 compliant database connection
                or cursor.
            query (str | Iterator): A query that will be rendered with the given data.
            data (Optional[Mapping]): A data mapping that will be rendered as params
                with the query. Optional, but required if the query contains parameters.
            Constructor (class): A constructor to use to build records from the results.
                The constructor must take the results of `zip(keys, values)` as its
                argument.

        Yields:
            record (Mapping): A mapping object that contains a database record.
        """
        cursor = self.execute(connection, query, data)
        fields = [d[0] for d in cursor.description]
        for row in cursor:
            yield Constructor(**dict(zip(fields, row)))

    def select_one(
        self,
        connection: Any,
        query: str | Iterator,
        data: Optional[Mapping] = None,
        Constructor=dict,
    ):
        records = self.select(
            connection=connection, query=query, data=data, Constructor=Constructor
        )
        for record in records:
            return record

    def select_all(
        self,
        connection: Any,
        query: str | Iterator,
        data: Optional[Mapping] = None,
        Constructor=dict,
    ):
        return list(
            self.select(
                connection=connection, query=query, data=data, Constructor=Constructor
            )
        )

Q

Convenience methods for building dynamic queries.

Examples:

>>> d = {"name": "Cheeseshop"}
>>> f"INSERT INTO tablename ({Q.fields(d)}) VALUES ({Q.params(d)})"
'INSERT INTO tablename (name) VALUES (:name)'
>>> f"SELECT ({Q.fields(d)}) FROM tablename WHERE {Q.filter('name')}"
'SELECT (name) FROM tablename WHERE name = :name'
>>> " ".join([
...     "UPDATE tablename SET",
...     Q.assigns(['name']),
...     "WHERE",
...     Q.filter('id'),
... ])
'UPDATE tablename SET name = :name WHERE id = :id'
>>> f"DELETE FROM tablename WHERE {Q.filter('name')}"
'DELETE FROM tablename WHERE name = :name'
Source code in sqly/sql.py 
class Q:
    """
    Convenience methods for building dynamic queries.

    Examples:
        >>> d = {"name": "Cheeseshop"}
        >>> f"INSERT INTO tablename ({Q.fields(d)}) VALUES ({Q.params(d)})"
        'INSERT INTO tablename (name) VALUES (:name)'
        >>> f"SELECT ({Q.fields(d)}) FROM tablename WHERE {Q.filter('name')}"
        'SELECT (name) FROM tablename WHERE name = :name'
        >>> " ".join([
        ...     "UPDATE tablename SET",
        ...     Q.assigns(['name']),
        ...     "WHERE",
        ...     Q.filter('id'),
        ... ])
        'UPDATE tablename SET name = :name WHERE id = :id'
        >>> f"DELETE FROM tablename WHERE {Q.filter('name')}"
        'DELETE FROM tablename WHERE name = :name'
    """

    @classmethod
    def keys(cls, fields: Iterable) -> list:
        """
        Return a list of field names from the given iterator.

        Arguments:
            fields (Iterable): An iterable of field names. (Can be a Mapping with field
                names as keys.)

        Returns:
            (list): A list of field names

        Examples:
            >>> Q.keys({'id': 1, 'name': 'Mark'})
            ['id', 'name']
        """
        return list(fields)

    @classmethod
    def fields(cls, fields: Iterable) -> str:
        """
        Render a comma-separated string of field names from the given fields. Use: E.g.,
        for dynamically specifying SELECT or INSERT field lists.

        Arguments:
            fields (Iterable): An iterable of field names. (Can be a Mapping with field
                names as keys.)

        Returns:
            (str): A comma-separated string of field names

        Examples:
            >>> Q.fields({'id': 1, 'name': 'Mark'})
            'id, name'
        """
        return ", ".join(fields)

    @classmethod
    def params(cls, fields: Iterable) -> str:
        """
        Render a comma-separated list of parameters from the given fields. Use: E.g.,
        dynamically specifying INSERT parameter lists.

        Arguments:
            fields (Iterable): An iterable of field names. (Can be a Mapping with field
                names as keys.)

        Returns:
            (str): A comma-separated string of field names

        Examples:
            >>> Q.params({'id': 1, 'name': 'Mark'})
            ':id, :name'
        """
        return ", ".join(f":{key}" for key in cls.keys(fields))

    @classmethod
    def assigns(cls, fields: Iterable) -> str:
        """
        Render a comma-separated list of assignments from the given fields. Use: E.g.,
        for dynamically specifying UPDATE field lists.

        Arguments:
            fields (Iterable): An iterable of field names. (Can be a Mapping with field
                names as keys.)

        Returns:
            (str): A comma-separated string of field `key = :key` assignments

        Examples:
            >>> Q.assigns({'id': 1, 'name': 'Mark'})
            'id = :id, name = :name'
        """
        return ", ".join(f"{key} = :{key}" for key in cls.keys(fields))

    @classmethod
    def filter(cls, field: str, *, op: Optional[str] = "="):
        """
        Render a filter from the given field, optional operator, and optional value.

        Arguments:
            field (str): The name of the field.
            op (str): The operator to use in the filter.

        Returns:
            (str): A comma-separated string of field names

        Examples:
            >>> Q.filter('id', op='>')
            'id > :id'
        """
        return f"{field} {op} :{field}"

    @classmethod
    def ANY(cls, field):
        """
        Use ANY(...) for list parameters
        """
        return f"{field} = ANY(:{field})"

ANY(field) classmethod

Use ANY(…) for list parameters

Source code in sqly/sql.py 
@classmethod
def ANY(cls, field):
    """
    Use ANY(...) for list parameters
    """
    return f"{field} = ANY(:{field})"

assigns(fields) classmethod

Render a comma-separated list of assignments from the given fields. Use: E.g., for dynamically specifying UPDATE field lists.

Parameters:

Name Type Description Default
fields Iterable

An iterable of field names. (Can be a Mapping with field names as keys.)

 required

Returns:

Type Description
(str)

A comma-separated string of field key = :key assignments

Examples:

>>> Q.assigns({'id': 1, 'name': 'Mark'})
'id = :id, name = :name'
Source code in sqly/sql.py 
@classmethod
def assigns(cls, fields: Iterable) -> str:
    """
    Render a comma-separated list of assignments from the given fields. Use: E.g.,
    for dynamically specifying UPDATE field lists.

    Arguments:
        fields (Iterable): An iterable of field names. (Can be a Mapping with field
            names as keys.)

    Returns:
        (str): A comma-separated string of field `key = :key` assignments

    Examples:
        >>> Q.assigns({'id': 1, 'name': 'Mark'})
        'id = :id, name = :name'
    """
    return ", ".join(f"{key} = :{key}" for key in cls.keys(fields))

fields(fields) classmethod

Render a comma-separated string of field names from the given fields. Use: E.g., for dynamically specifying SELECT or INSERT field lists.

Parameters:

Name Type Description Default
fields Iterable

An iterable of field names. (Can be a Mapping with field names as keys.)

 required

Returns:

Type Description
(str)

A comma-separated string of field names

Examples:

>>> Q.fields({'id': 1, 'name': 'Mark'})
'id, name'
Source code in sqly/sql.py 
@classmethod
def fields(cls, fields: Iterable) -> str:
    """
    Render a comma-separated string of field names from the given fields. Use: E.g.,
    for dynamically specifying SELECT or INSERT field lists.

    Arguments:
        fields (Iterable): An iterable of field names. (Can be a Mapping with field
            names as keys.)

    Returns:
        (str): A comma-separated string of field names

    Examples:
        >>> Q.fields({'id': 1, 'name': 'Mark'})
        'id, name'
    """
    return ", ".join(fields)

filter(field, *, op='=') classmethod

Render a filter from the given field, optional operator, and optional value.

Parameters:

Name Type Description Default
field str

The name of the field.

 required
op str

The operator to use in the filter.

 '='

Returns:

Type Description
(str)

A comma-separated string of field names

Examples:

>>> Q.filter('id', op='>')
'id > :id'
Source code in sqly/sql.py 
@classmethod
def filter(cls, field: str, *, op: Optional[str] = "="):
    """
    Render a filter from the given field, optional operator, and optional value.

    Arguments:
        field (str): The name of the field.
        op (str): The operator to use in the filter.

    Returns:
        (str): A comma-separated string of field names

    Examples:
        >>> Q.filter('id', op='>')
        'id > :id'
    """
    return f"{field} {op} :{field}"

keys(fields) classmethod

Return a list of field names from the given iterator.

Parameters:

Name Type Description Default
fields Iterable

An iterable of field names. (Can be a Mapping with field names as keys.)

 required

Returns:

Type Description
(list)

A list of field names

Examples:

>>> Q.keys({'id': 1, 'name': 'Mark'})
['id', 'name']
Source code in sqly/sql.py 
@classmethod
def keys(cls, fields: Iterable) -> list:
    """
    Return a list of field names from the given iterator.

    Arguments:
        fields (Iterable): An iterable of field names. (Can be a Mapping with field
            names as keys.)

    Returns:
        (list): A list of field names

    Examples:
        >>> Q.keys({'id': 1, 'name': 'Mark'})
        ['id', 'name']
    """
    return list(fields)

params(fields) classmethod

Render a comma-separated list of parameters from the given fields. Use: E.g., dynamically specifying INSERT parameter lists.

Parameters:

Name Type Description Default
fields Iterable

An iterable of field names. (Can be a Mapping with field names as keys.)

 required

Returns:

Type Description
(str)

A comma-separated string of field names

Examples:

>>> Q.params({'id': 1, 'name': 'Mark'})
':id, :name'
Source code in sqly/sql.py 
@classmethod
def params(cls, fields: Iterable) -> str:
    """
    Render a comma-separated list of parameters from the given fields. Use: E.g.,
    dynamically specifying INSERT parameter lists.

    Arguments:
        fields (Iterable): An iterable of field names. (Can be a Mapping with field
            names as keys.)

    Returns:
        (str): A comma-separated string of field names

    Examples:
        >>> Q.params({'id': 1, 'name': 'Mark'})
        ':id, :name'
    """
    return ", ".join(f":{key}" for key in cls.keys(fields))

execute(self, connection, query, data=None)

Execute the given query on the connection and return the connection cursor.

If the query fails: Rollback the connection and re-raise the exception, as a convenience to the user not to leave the connection in an unusable state.

If .execute() is called with a previously-generated cursor, that cursor will be reused and the same cursor returned from the method call.

Parameters:

Name Type Description Default
connection Connection | Cursor

A DB-API 2.0 compliant database connection or cursor.

 required
query str | Iterator

A query that will be rendered with the given data.

 required
data Optional[Mapping]

A data mapping that will be rendered as params with the query. Optional, but required if the query contains parameters.

 None

Returns:

Type Description
cursor (Cursor)

A DB-API 2.0 compliant database cursor.
Source code in sqly/sql.py 
def execute(
    self, connection: Any, query: str | Iterator, data: Optional[Mapping] = None
):
    """
    Execute the given query on the connection and return the connection cursor.

    If the query fails: Rollback the connection and re-raise the exception, as a
    convenience to the user not to leave the connection in an unusable state.

    If `.execute()` is called with a previously-generated cursor, that cursor will
    be reused and the same cursor returned from the method call.

    Parameters:
        connection (Connection | Cursor): A DB-API 2.0 compliant database connection
            or cursor.
        query (str | Iterator): A query that will be rendered with the given data.
        data (Optional[Mapping]): A data mapping that will be rendered as params
            with the query. Optional, but required if the query contains parameters.

    Returns:
        cursor (Cursor): A DB-API 2.0 compliant database cursor.
    """
    try:
        cursor = connection.execute(*self.render(query, data))
    except Exception as exc:
        # If the connection is a cursor, get the underlying connection to rollback,
        # because cursors don't have a rollback method.
        if hasattr(connection, "rollback"):
            try:
                connection.rollback()
            except Exception:
                ...
        raise exc

    return cursor

render(self, query, data=None)

Render a query string and its parameters for this SQL dialect.

Parameters:

Name Type Description Default
query str | Iterator

a string or iterator of strings.

 required
data Mapping

a keyword dict used to render the query parameters.

 None

Returns:

Type Description
(str)

the rendered query string. (tuple | dict): depends on the param format:

- positional param formats (QMARK, NUMBERED) return a tuple of values
- named param formats (NAMED, PYFORMAT) return a dict
Source code in sqly/sql.py 
def render(self, query, data=None):
    """
    Render a query string and its parameters for this SQL dialect.

    Arguments:
        query (str | Iterator): a string or iterator of strings.
        data (Mapping): a keyword dict used to render the query parameters.

    Returns:
        (str): the rendered query string.
        (tuple | dict): depends on the param format:

            - positional param formats (QMARK, NUMBERED) return a tuple of values
            - named param formats (NAMED, PYFORMAT) return a dict
    """
    # ordered list of fields for positional outputs (closure for replace_parameter)
    fields = []

    def replace_parameter(match):
        field = match.group(1)

        # Build the ordered fields list
        if self.dialect.param_format.is_positional or field not in fields:
            fields.append(field)

        # Return the field formatted for the param format type
        if self.dialect.param_format == ParamFormat.NAMED:
            return f":{field}"
        elif self.dialect.param_format == ParamFormat.PYFORMAT:
            return f"%({field})s"
        elif self.dialect.param_format == ParamFormat.QMARK:
            return "?"
        elif self.dialect.param_format == ParamFormat.NUMBERED:
            return f"${len(fields)}"
        else:  # self.dialect.param_format == ParamFormat.FORMAT:
            return "%s"

    # 1. Convert query to a string
    if isinstance(query, str):
        query_str = str(query)
    elif hasattr(query, "__iter__"):
        query_str = "\n".join(str(q) for q in walk(query))
    else:
        raise ValueError(f"Query has unsupported type: {type(query)}")

    # 2. Escape string parameters in the PYFORMAT param format
    if self.dialect.param_format == ParamFormat.PYFORMAT:
        # any % must be intended as literal and must be doubled
        query_str = query_str.replace("%", "%%")

    # 3. Replace the parameter with its dialect-specific representation
    pattern = r"(?<!\\):(\w+)\b"  # colon + word not preceded by a backslash
    query_str = re.sub(pattern, replace_parameter, query_str).strip()

    # 4. Un-escape remaining escaped colon params
    if self.dialect.param_format == ParamFormat.NAMED:
        # replace \:word with :word because the colon-escape is no longer needed.
        query_str = re.sub(r"\\:(\w+)\b", r":\1", query_str)

    # 5. Build the parameter_values dict or list for use with the query
    if self.dialect.param_format.is_positional:
        # parameter_values is a list of values
        parameter_values = [
            json.dumps(val) if isinstance(val, dict) else val
            for val in [data[field] for field in fields]
        ]
    else:
        # parameter_values is a dict of key:value fields
        parameter_values = {
            # (dict, set, tuple) for json/b, but list is for "IN / ANY()" params.
            key: json.dumps(val) if isinstance(val, (dict, set, tuple)) else val
            for key, val in {field: data[field] for field in fields}.items()
        }

    # 6. Return a tuple formatted for this Dialect
    if self.dialect == Dialect.ASYNCPG:
        # asyncpg expects the parameters in a tuple following the query string.
        return tuple([query_str] + parameter_values)
    else:
        # other dialects expect the parameters in the second tuple item.
        return (query_str, parameter_values)

select(self, connection, query, data=None, Constructor=<class 'dict'>)

Execute the given query on the connection, and yield result records.

The .select() method is a generator which iterates over a native database cursor. The results of the method can be cast to a list via list(...) or can be iterated through one at a time.

If the query fails: Rollback the connection and re-raise the exception, as a convenience to the user not to leave the connection in an unusable state.

Parameters:

Name Type Description Default
connection Connection | Cursor

A DB-API 2.0 compliant database connection or cursor.

 required
query str | Iterator

A query that will be rendered with the given data.

 required
data Optional[Mapping]

A data mapping that will be rendered as params with the query. Optional, but required if the query contains parameters.

 None
Constructor class

A constructor to use to build records from the results. The constructor must take the results of zip(keys, values) as its argument.

 <class 'dict'>

Yields:

Type Description
record (Mapping)

A mapping object that contains a database record.
Source code in sqly/sql.py 
def select(
    self,
    connection: Any,
    query: str | Iterator,
    data: Optional[Mapping] = None,
    Constructor=dict,
):
    """
    Execute the given query on the connection, and yield result records.

    The `.select()` method is a generator which iterates over a native database
    cursor. The results of the method can be cast to a list via `list(...)` or can
    be iterated through one at a time.

    If the query fails: Rollback the connection and re-raise the exception, as a
    convenience to the user not to leave the connection in an unusable state.

    Parameters:
        connection (Connection | Cursor): A DB-API 2.0 compliant database connection
            or cursor.
        query (str | Iterator): A query that will be rendered with the given data.
        data (Optional[Mapping]): A data mapping that will be rendered as params
            with the query. Optional, but required if the query contains parameters.
        Constructor (class): A constructor to use to build records from the results.
            The constructor must take the results of `zip(keys, values)` as its
            argument.

    Yields:
        record (Mapping): A mapping object that contains a database record.
    """
    cursor = self.execute(connection, query, data)
    fields = [d[0] for d in cursor.description]
    for row in cursor:
        yield Constructor(**dict(zip(fields, row)))