Skip to content

turu.mysql

turu.mysql.Connection

Source code in turu-mysql/src/turu/mysql/connection.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class Connection(turu.core.connection.Connection):
    def __init__(self, connection: pymysql.Connection) -> None:
        self._raw_connection = connection

    @override
    @classmethod
    def connect(  # type: ignore[override]
        cls,
        user: Optional[str] = None,
        password: str = "",
        host: Optional[str] = None,
        database: Optional[str] = None,
        port: int = 0,
        **kwargs: Unpack["_ConnectParams"],
    ) -> Self:
        return cls(
            pymysql.connect(
                user=user,
                password=password,
                host=host,
                database=database,
                port=port,
                **kwargs,
            )
        )

    @override
    @classmethod
    def connect_from_env(  # type: ignore[override]
        cls,
        user_envname: str = "MYSQL_USER",
        password_envname: str = "MYSQL_PASSWORD",
        host_envname: str = "MYSQL_HOST",
        database_envname: str = "MYSQL_DATABASE",
        port_envname: str = "MYSQL_PORT",
        **kwargs: Unpack["_ConnectParams"],
    ) -> Self:
        return cls.connect(
            user=os.environ.get(user_envname),
            password=os.environ.get(password_envname, ""),
            host=os.environ.get(host_envname),
            database=os.environ.get(database_envname),
            port=int(os.environ.get(port_envname, 0)),
            **kwargs,
        )

    @override
    def close(self) -> None:
        self._raw_connection.close()

    @override
    def commit(self) -> None:
        self._raw_connection.commit()

    @override
    def rollback(self) -> None:
        self._raw_connection.rollback()

    @override
    def cursor(self) -> "Cursor[Never]":
        return Cursor(self._raw_connection.cursor())

close()

Close the connection now.

Source code in turu-mysql/src/turu/mysql/connection.py 
59
60
61
@override
def close(self) -> None:
    self._raw_connection.close()

commit()

Commit any pending transaction to the database.

Source code in turu-mysql/src/turu/mysql/connection.py 
63
64
65
@override
def commit(self) -> None:
    self._raw_connection.commit()

connect(user=None, password='', host=None, database=None, port=0, **kwargs) classmethod

Connect to a database.

Source code in turu-mysql/src/turu/mysql/connection.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@override
@classmethod
def connect(  # type: ignore[override]
    cls,
    user: Optional[str] = None,
    password: str = "",
    host: Optional[str] = None,
    database: Optional[str] = None,
    port: int = 0,
    **kwargs: Unpack["_ConnectParams"],
) -> Self:
    return cls(
        pymysql.connect(
            user=user,
            password=password,
            host=host,
            database=database,
            port=port,
            **kwargs,
        )
    )

connect_from_env(user_envname='MYSQL_USER', password_envname='MYSQL_PASSWORD', host_envname='MYSQL_HOST', database_envname='MYSQL_DATABASE', port_envname='MYSQL_PORT', **kwargs) classmethod

Connect to a database using environment variables.

Source code in turu-mysql/src/turu/mysql/connection.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@override
@classmethod
def connect_from_env(  # type: ignore[override]
    cls,
    user_envname: str = "MYSQL_USER",
    password_envname: str = "MYSQL_PASSWORD",
    host_envname: str = "MYSQL_HOST",
    database_envname: str = "MYSQL_DATABASE",
    port_envname: str = "MYSQL_PORT",
    **kwargs: Unpack["_ConnectParams"],
) -> Self:
    return cls.connect(
        user=os.environ.get(user_envname),
        password=os.environ.get(password_envname, ""),
        host=os.environ.get(host_envname),
        database=os.environ.get(database_envname),
        port=int(os.environ.get(port_envname, 0)),
        **kwargs,
    )

cursor()

Return a new Cursor Object using the connection.

Source code in turu-mysql/src/turu/mysql/connection.py 
71
72
73
@override
def cursor(self) -> "Cursor[Never]":
    return Cursor(self._raw_connection.cursor())

rollback()

Roll back to the start of any pending transaction.

Closing a connection without committing the changes first will cause an implicit rollback to be performed.

Source code in turu-mysql/src/turu/mysql/connection.py 
67
68
69
@override
def rollback(self) -> None:
    self._raw_connection.rollback()

turu.mysql.Cursor

Source code in turu-mysql/src/turu/mysql/cursor.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class Cursor(
    turu.core.cursor.Cursor[turu.core.cursor.GenericRowType, Parameters],
):
    def __init__(
        self,
        cursor: pymysql.cursors.Cursor,
        *,
        row_type: Optional[Type[turu.core.cursor.GenericRowType]] = None,
    ):
        self._raw_cursor = cursor
        self._row_type: Optional[Type[turu.core.cursor.GenericRowType]] = row_type
        self._iter = None

    @property
    def rowcount(self) -> int:
        return self._raw_cursor.rowcount

    @property
    def arraysize(self) -> int:
        return self._raw_cursor.arraysize

    @arraysize.setter
    def arraysize(self, size: int) -> None:
        self._raw_cursor.arraysize = size

    @override
    def close(self) -> None:
        self._raw_cursor.close()

    @override
    def execute(
        self,
        operation: str,
        parameters: Optional[Parameters] = None,
        /,
    ) -> "Cursor[turu.core.cursor.GenericRowType]":
        self._raw_cursor.execute(cast(LiteralString, operation), parameters)
        self._row_type = None

        return self

    @override
    def executemany(
        self,
        operation: str,
        seq_of_parameters: Sequence[Parameters],
        /,
    ) -> "Cursor[turu.core.cursor.GenericRowType]":
        self._raw_cursor.executemany(cast(LiteralString, operation), seq_of_parameters)
        self._row_type = None

        return self

    @override
    def execute_map(
        self,
        row_type: Type[turu.core.cursor.GenericNewRowType],
        operation: str,
        parameters: Optional[Parameters] = None,
        /,
    ) -> "Cursor[turu.core.cursor.GenericNewRowType]":
        self._raw_cursor.execute(cast(LiteralString, operation), parameters)
        self._row_type = cast(Type[turu.core.cursor.GenericRowType], row_type)

        return self  # type: ignore

    @override
    def executemany_map(
        self,
        row_type: Type[turu.core.cursor.GenericNewRowType],
        operation: str,
        seq_of_parameters: Sequence[Parameters],
        /,
    ) -> "Cursor[turu.core.cursor.GenericNewRowType]":
        self._raw_cursor.executemany(cast(LiteralString, operation), seq_of_parameters)
        self._row_type = cast(Type[turu.core.cursor.GenericRowType], row_type)

        return cast(Cursor, self)

    @override
    def execute_with_tag(
        self,
        tag: Type[turu.core.tag.Tag],
        operation: str,
        parameters: "Optional[Parameters]" = None,
    ) -> "Cursor[Never]":
        return cast(Cursor, self.execute(operation, parameters))

    @override
    def executemany_with_tag(
        self,
        tag: Type[turu.core.tag.Tag],
        operation: str,
        seq_of_parameters: "Sequence[Parameters]",
    ) -> "Cursor[Never]":
        return cast(Cursor, self.executemany(operation, seq_of_parameters))

    @override
    def fetchone(self) -> Optional[turu.core.cursor.GenericRowType]:
        row = self._raw_cursor.fetchone()
        if row is None:
            return None

        elif self._row_type is not None:
            return turu.core.cursor.map_row(self._row_type, row)

        else:
            return row  # type: ignore

    @override
    def fetchmany(
        self, size: Optional[int] = None
    ) -> List[turu.core.cursor.GenericRowType]:
        return [
            turu.core.cursor.map_row(self._row_type, row)
            for row in self._raw_cursor.fetchmany(
                size if size is not None else self.arraysize
            )
        ]

    @override
    def fetchall(self) -> List[turu.core.cursor.GenericRowType]:
        return [
            turu.core.cursor.map_row(self._row_type, row)
            for row in self._raw_cursor.fetchall()
        ]

    @override
    def __iter__(self) -> "Cursor[turu.core.cursor.GenericRowType]":
        self._iter = self._raw_cursor.__iter__()
        return self

    @override
    def __next__(self) -> turu.core.cursor.GenericRowType:
        if self._iter is None:
            self._iter = self._raw_cursor.__iter__()

        next_row = next(self._iter)
        if self._row_type is not None and next_row is not None:
            return turu.core.cursor.map_row(self._row_type, next_row)

        else:
            return next_row  # type: ignore

arraysize: int property writable

The number of rows to fetch at a time with .fetchmany().

It defaults to 1 meaning to fetch a single row at a time.

rowcount: int property

The number of rows that the last .execute*() produced (for DQL statements like ) or affected (for DML statements like or ).

The attribute is -1 in case no .execute*() has been performed on the cursor or the rowcount of the last operation is cannot be determined by the interface.

execute(operation, parameters=None)

Prepare and execute a database operation (query or command).

Parameters:

Name Type Description Default
operation str

A database operation (query or command).

 required
parameters Optional[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 None

Returns:

Type Description
Self

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/cursor.py 
41
42
43
44
45
46
47
48
49
50
51
@override
def execute(
    self,
    operation: str,
    parameters: Optional[Parameters] = None,
    /,
) -> "Cursor[turu.core.cursor.GenericRowType]":
    self._raw_cursor.execute(cast(LiteralString, operation), parameters)
    self._row_type = None

    return self

execute_map(row_type, operation, parameters=None)

Execute a database operation (query or command) and map each row to a row_type.

Parameters:

Name Type Description Default
row_type Type[GenericNewRowType]

The type of the row that will be returned.

 required
operation str

A database operation (query or command).

 required
parameters Optional[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 None

Returns:

Type Description
Cursor[GenericNewRowType, Parameters]

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/cursor.py 
65
66
67
68
69
70
71
72
73
74
75
76
@override
def execute_map(
    self,
    row_type: Type[turu.core.cursor.GenericNewRowType],
    operation: str,
    parameters: Optional[Parameters] = None,
    /,
) -> "Cursor[turu.core.cursor.GenericNewRowType]":
    self._raw_cursor.execute(cast(LiteralString, operation), parameters)
    self._row_type = cast(Type[turu.core.cursor.GenericRowType], row_type)

    return self  # type: ignore

execute_with_tag(tag, operation, parameters=None)

Execute a database operation (Insert, Update, Delete) with a tag.

This is not defined in PEP 249,

This method is provided for testing, and is intended to be used in conjunction with MockConnection.inject_operation_with_tag.

Source code in turu-mysql/src/turu/mysql/cursor.py 
91
92
93
94
95
96
97
98
@override
def execute_with_tag(
    self,
    tag: Type[turu.core.tag.Tag],
    operation: str,
    parameters: "Optional[Parameters]" = None,
) -> "Cursor[Never]":
    return cast(Cursor, self.execute(operation, parameters))

executemany(operation, seq_of_parameters)

Prepare a database operation (query or command) and then execute it against all parameter sequences or mappings.

Parameters:

Name Type Description Default
operation str

A database operation (query or command).

 required
seq_of_parameters Sequence[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 required

Returns:

Type Description
Self

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/cursor.py 
53
54
55
56
57
58
59
60
61
62
63
@override
def executemany(
    self,
    operation: str,
    seq_of_parameters: Sequence[Parameters],
    /,
) -> "Cursor[turu.core.cursor.GenericRowType]":
    self._raw_cursor.executemany(cast(LiteralString, operation), seq_of_parameters)
    self._row_type = None

    return self

executemany_map(row_type, operation, seq_of_parameters)

Execute a database operation (query or command) against all parameter sequences or mappings.

Parameters:

Name Type Description Default
row_type Type[GenericNewRowType]

The type of the row that will be returned.

 required
operation str

A database operation (query or command).

 required
seq_of_parameters Sequence[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 required

Returns:

Type Description
Cursor[GenericNewRowType, Parameters]

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/cursor.py 
78
79
80
81
82
83
84
85
86
87
88
89
@override
def executemany_map(
    self,
    row_type: Type[turu.core.cursor.GenericNewRowType],
    operation: str,
    seq_of_parameters: Sequence[Parameters],
    /,
) -> "Cursor[turu.core.cursor.GenericNewRowType]":
    self._raw_cursor.executemany(cast(LiteralString, operation), seq_of_parameters)
    self._row_type = cast(Type[turu.core.cursor.GenericRowType], row_type)

    return cast(Cursor, self)

executemany_with_tag(tag, operation, seq_of_parameters)

Execute a database operation (Insert, Update, Delete) against all parameter sequences or mappings with a tag.

This is not defined in PEP 249,

This method is provided for testing, and is intended to be used in conjunction with MockConnection.inject_operation_with_tag.

Source code in turu-mysql/src/turu/mysql/cursor.py 
100
101
102
103
104
105
106
107
@override
def executemany_with_tag(
    self,
    tag: Type[turu.core.tag.Tag],
    operation: str,
    seq_of_parameters: "Sequence[Parameters]",
) -> "Cursor[Never]":
    return cast(Cursor, self.executemany(operation, seq_of_parameters))

fetchall()

Fetch all (remaining) rows of a query result

Source code in turu-mysql/src/turu/mysql/cursor.py 
132
133
134
135
136
137
@override
def fetchall(self) -> List[turu.core.cursor.GenericRowType]:
    return [
        turu.core.cursor.map_row(self._row_type, row)
        for row in self._raw_cursor.fetchall()
    ]

fetchmany(size=None)

Fetch the next set of rows of a query result.

An empty sequence is returned when no more rows are available.

Parameters:

Name Type Description Default
size Optional[int]

The number of rows to fetch per call. If this parameter is not used, it is usually refer to use the .arraysize attribute (Default is 1). If this parameter is used, then it is best for it to retain the same value from one .fetchmany() call to the next.

 None
Source code in turu-mysql/src/turu/mysql/cursor.py 
121
122
123
124
125
126
127
128
129
130
@override
def fetchmany(
    self, size: Optional[int] = None
) -> List[turu.core.cursor.GenericRowType]:
    return [
        turu.core.cursor.map_row(self._row_type, row)
        for row in self._raw_cursor.fetchmany(
            size if size is not None else self.arraysize
        )
    ]

fetchone()

Fetch the next row of a query result set.

Source code in turu-mysql/src/turu/mysql/cursor.py 
109
110
111
112
113
114
115
116
117
118
119
@override
def fetchone(self) -> Optional[turu.core.cursor.GenericRowType]:
    row = self._raw_cursor.fetchone()
    if row is None:
        return None

    elif self._row_type is not None:
        return turu.core.cursor.map_row(self._row_type, row)

    else:
        return row  # type: ignore

turu.mysql.AsyncConnection

Source code in turu-mysql/src/turu/mysql/async_connection.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class AsyncConnection(turu.core.async_connection.AsyncConnection):
    def __init__(self, connection: aiomysql.Connection):
        self._raw_connection = connection

    @override
    @classmethod
    async def connect(  # type: ignore[override]
        cls,
        user: Optional[str] = None,
        password: str = "",
        host: str = "localhost",
        database: Optional[str] = None,
        port: int = 0,
        **kwargs: Unpack["_ConnectParams"],
    ) -> Self:
        return cls(
            await aiomysql.connection._connect(
                user=user,
                password=password,
                host=host,
                db=database,
                port=port,
                **kwargs,
            )
        )

    @override
    @classmethod
    async def connect_from_env(  # type: ignore[override]
        cls,
        user_envname: str = "MYSQL_USER",
        password_envname: str = "MYSQL_PASSWORD",
        host_envname: str = "MYSQL_HOST",
        database_envname: str = "MYSQL_DATABASE",
        port_envname: str = "MYSQL_PORT",
        **kwargs: Unpack["_ConnectParams"],
    ) -> Self:
        return await cls.connect(
            user=os.environ.get(user_envname),
            password=os.environ.get(password_envname, ""),
            host=os.environ.get(host_envname, "localhost"),
            database=os.environ.get(database_envname),
            port=int(os.environ.get(port_envname, 0)),
            **kwargs,
        )

    @override
    async def close(self) -> None:
        await self._raw_connection.ensure_closed()

    @override
    async def commit(self) -> None:
        await self._raw_connection.commit()

    @override
    async def rollback(self) -> None:
        await self._raw_connection.rollback()

    @override
    async def cursor(self) -> AsyncCursor[Never]:
        return AsyncCursor(
            aiomysql.cursors.Cursor(
                self._raw_connection,
                self._raw_connection._echo,
            )
        )

close() async

Close the connection now.

Source code in turu-mysql/src/turu/mysql/async_connection.py 
63
64
65
@override
async def close(self) -> None:
    await self._raw_connection.ensure_closed()

commit() async

Commit any pending transaction to the database.

Source code in turu-mysql/src/turu/mysql/async_connection.py 
67
68
69
@override
async def commit(self) -> None:
    await self._raw_connection.commit()

connect(user=None, password='', host='localhost', database=None, port=0, **kwargs) async classmethod

Connect to a database.

Source code in turu-mysql/src/turu/mysql/async_connection.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@override
@classmethod
async def connect(  # type: ignore[override]
    cls,
    user: Optional[str] = None,
    password: str = "",
    host: str = "localhost",
    database: Optional[str] = None,
    port: int = 0,
    **kwargs: Unpack["_ConnectParams"],
) -> Self:
    return cls(
        await aiomysql.connection._connect(
            user=user,
            password=password,
            host=host,
            db=database,
            port=port,
            **kwargs,
        )
    )

connect_from_env(user_envname='MYSQL_USER', password_envname='MYSQL_PASSWORD', host_envname='MYSQL_HOST', database_envname='MYSQL_DATABASE', port_envname='MYSQL_PORT', **kwargs) async classmethod

Connect to a database using environment variables.

Source code in turu-mysql/src/turu/mysql/async_connection.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@override
@classmethod
async def connect_from_env(  # type: ignore[override]
    cls,
    user_envname: str = "MYSQL_USER",
    password_envname: str = "MYSQL_PASSWORD",
    host_envname: str = "MYSQL_HOST",
    database_envname: str = "MYSQL_DATABASE",
    port_envname: str = "MYSQL_PORT",
    **kwargs: Unpack["_ConnectParams"],
) -> Self:
    return await cls.connect(
        user=os.environ.get(user_envname),
        password=os.environ.get(password_envname, ""),
        host=os.environ.get(host_envname, "localhost"),
        database=os.environ.get(database_envname),
        port=int(os.environ.get(port_envname, 0)),
        **kwargs,
    )

cursor() async

Return a new Cursor Object using the connection.

Source code in turu-mysql/src/turu/mysql/async_connection.py 
75
76
77
78
79
80
81
82
@override
async def cursor(self) -> AsyncCursor[Never]:
    return AsyncCursor(
        aiomysql.cursors.Cursor(
            self._raw_connection,
            self._raw_connection._echo,
        )
    )

rollback() async

Roll back to the start of any pending transaction.

Closing a connection without committing the changes first will cause an implicit rollback to be performed.

Source code in turu-mysql/src/turu/mysql/async_connection.py 
71
72
73
@override
async def rollback(self) -> None:
    await self._raw_connection.rollback()

turu.mysql.AsyncCursor

Source code in turu-mysql/src/turu/mysql/async_cursor.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class AsyncCursor(
    turu.core.async_cursor.AsyncCursor[
        turu.core.async_cursor.GenericRowType, Parameters
    ],
):
    def __init__(
        self,
        cursor: aiomysql.Cursor,
        *,
        row_type: Optional[Type[turu.core.async_cursor.GenericRowType]] = None,
    ):
        self._raw_cursor = cursor
        self._row_type: Optional[Type[turu.core.async_cursor.GenericRowType]] = row_type
        self._aiter = None

    @property
    def rowcount(self) -> int:
        return self._raw_cursor.rowcount

    @property
    def arraysize(self) -> int:
        return self._raw_cursor.arraysize

    @arraysize.setter
    def arraysize(self, size: int) -> None:
        self._raw_cursor.arraysize = size

    @override
    async def close(self) -> None:
        await self._raw_cursor.close()

    @override
    async def execute(
        self, operation: str, parameters: Optional[Parameters] = None, /
    ) -> "AsyncCursor[Tuple[Any]]":
        await self._raw_cursor.execute(cast(LiteralString, operation), parameters)
        self._row_type = None

        return cast(AsyncCursor, self)

    @override
    async def executemany(
        self, operation: str, seq_of_parameters: Sequence[Parameters], /
    ) -> "AsyncCursor[Tuple[Any]]":
        await self._raw_cursor.executemany(
            cast(LiteralString, operation), seq_of_parameters
        )
        self._row_type = None

        return cast(AsyncCursor, self)

    @override
    async def execute_map(
        self,
        row_type: Type[turu.core.async_cursor.GenericNewRowType],
        operation: str,
        parameters: Optional[Parameters] = None,
        /,
    ) -> "AsyncCursor[turu.core.async_cursor.GenericNewRowType]":
        await self._raw_cursor.execute(cast(LiteralString, operation), parameters)
        self._row_type = cast(Type[turu.core.async_cursor.GenericRowType], row_type)

        return cast(AsyncCursor, self)

    @override
    async def executemany_map(
        self,
        row_type: Type[turu.core.async_cursor.GenericNewRowType],
        operation: str,
        seq_of_parameters: Sequence[Parameters],
        /,
    ) -> "AsyncCursor[turu.core.async_cursor.GenericNewRowType]":
        await self._raw_cursor.executemany(
            cast(LiteralString, operation), seq_of_parameters
        )
        self._row_type = cast(Type[turu.core.async_cursor.GenericRowType], row_type)

        return cast(AsyncCursor, self)

    @override
    async def execute_with_tag(
        self,
        tag: Type[turu.core.tag.Tag],
        operation: str,
        parameters: "Optional[Parameters]" = None,
    ) -> turu.core.async_cursor.AsyncCursor[Never, Parameters]:
        return cast(
            turu.core.async_cursor.AsyncCursor,
            await self.execute(operation, parameters),
        )

    @override
    async def executemany_with_tag(
        self,
        tag: Type[turu.core.tag.Tag],
        operation: str,
        seq_of_parameters: Sequence[Parameters],
    ) -> turu.core.async_cursor.AsyncCursor[Never, Parameters]:
        return cast(
            turu.core.async_cursor.AsyncCursor,
            await self.executemany(operation, seq_of_parameters),
        )

    @override
    async def fetchone(self) -> Optional[turu.core.async_cursor.GenericRowType]:
        row = await self._raw_cursor.fetchone()
        if row is None:
            return None

        return _map_row(self._row_type, row)

    @override
    async def fetchmany(
        self, size: Optional[int] = None
    ) -> List[turu.core.async_cursor.GenericRowType]:
        return [
            _map_row(self._row_type, row)
            for row in (
                await self._raw_cursor.fetchmany(
                    size if size is not None else self.arraysize
                )
            )
        ]

    @override
    async def fetchall(self) -> List[turu.core.async_cursor.GenericRowType]:
        return [
            _map_row(self._row_type, row) for row in await self._raw_cursor.fetchall()
        ]

    @override
    def __aiter__(self) -> "AsyncCursor[turu.core.async_cursor.GenericRowType]":
        self._aiter = self._raw_cursor.__aiter__()
        return self

    @override
    async def __anext__(self) -> turu.core.async_cursor.GenericRowType:
        if self._aiter is None:
            self._aiter = self._raw_cursor.__aiter__()

        next_row = await self._aiter.__anext__()
        return _map_row(self._row_type, next_row)

arraysize: int property writable

This read/write attribute specifies the number of rows to fetch at a time with .fetchmany(). It defaults to 1 meaning to fetch a single row at a time.

Implementations must observe this value with respect to the .fetchmany() method, but are free to interact with the database a single row at a time. It may also be used in the implementation of .executemany().

rowcount: int property

the number of rows that the last .execute*() produced (for DQL statements like ) or affected (for DML statements like or ).

The attribute is -1 in case no .execute*() has been performed on the cursor or the rowcount of the last operation is cannot be determined by the interface.

execute(operation, parameters=None) async

Prepare and execute a database operation (query or command).

Parameters:

Name Type Description Default
operation str

A database operation (query or command).

 required
parameters Optional[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 None

Returns:

Type Description
Self

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/async_cursor.py 
52
53
54
55
56
57
58
59
@override
async def execute(
    self, operation: str, parameters: Optional[Parameters] = None, /
) -> "AsyncCursor[Tuple[Any]]":
    await self._raw_cursor.execute(cast(LiteralString, operation), parameters)
    self._row_type = None

    return cast(AsyncCursor, self)

execute_map(row_type, operation, parameters=None) async

Execute a database operation (query or command) and map each row to a row_type.

Parameters:

Name Type Description Default
row_type Type[GenericNewRowType]

The type of the row that will be returned.

 required
operation str

A database operation (query or command).

 required
parameters Optional[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 None

Returns:

Type Description
AsyncCursor[GenericNewRowType, Parameters]

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/async_cursor.py 
72
73
74
75
76
77
78
79
80
81
82
83
@override
async def execute_map(
    self,
    row_type: Type[turu.core.async_cursor.GenericNewRowType],
    operation: str,
    parameters: Optional[Parameters] = None,
    /,
) -> "AsyncCursor[turu.core.async_cursor.GenericNewRowType]":
    await self._raw_cursor.execute(cast(LiteralString, operation), parameters)
    self._row_type = cast(Type[turu.core.async_cursor.GenericRowType], row_type)

    return cast(AsyncCursor, self)

execute_with_tag(tag, operation, parameters=None) async

Execute a database operation (Insert, Update, Delete) with a tag.

This is not defined in PEP 249,

This method executes an operation (Insert, Update, Delete) that does not return a value with a tag. This tag is used to verify that the specified operation is executed in order when testing with Mock.

Source code in turu-mysql/src/turu/mysql/async_cursor.py 
100
101
102
103
104
105
106
107
108
109
110
@override
async def execute_with_tag(
    self,
    tag: Type[turu.core.tag.Tag],
    operation: str,
    parameters: "Optional[Parameters]" = None,
) -> turu.core.async_cursor.AsyncCursor[Never, Parameters]:
    return cast(
        turu.core.async_cursor.AsyncCursor,
        await self.execute(operation, parameters),
    )

executemany(operation, seq_of_parameters) async

Prepare a database operation (query or command) and then execute it against all parameter sequences or mappings.

Parameters:

Name Type Description Default
operation str

A database operation (query or command).

 required
seq_of_parameters Sequence[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 required

Returns:

Type Description
Self

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/async_cursor.py 
61
62
63
64
65
66
67
68
69
70
@override
async def executemany(
    self, operation: str, seq_of_parameters: Sequence[Parameters], /
) -> "AsyncCursor[Tuple[Any]]":
    await self._raw_cursor.executemany(
        cast(LiteralString, operation), seq_of_parameters
    )
    self._row_type = None

    return cast(AsyncCursor, self)

executemany_map(row_type, operation, seq_of_parameters) async

Execute a database operation (query or command) against all parameter sequences or mappings.

Parameters:

Name Type Description Default
row_type Type[GenericNewRowType]

The type of the row that will be returned.

 required
operation str

A database operation (query or command).

 required
seq_of_parameters Sequence[Parameters]

Parameters may be provided as sequence or mapping and will be bound to variables in the operation.

 required

Returns:

Type Description
AsyncCursor[GenericNewRowType, Parameters]

A cursor that holds a reference to an operation.
Source code in turu-mysql/src/turu/mysql/async_cursor.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@override
async def executemany_map(
    self,
    row_type: Type[turu.core.async_cursor.GenericNewRowType],
    operation: str,
    seq_of_parameters: Sequence[Parameters],
    /,
) -> "AsyncCursor[turu.core.async_cursor.GenericNewRowType]":
    await self._raw_cursor.executemany(
        cast(LiteralString, operation), seq_of_parameters
    )
    self._row_type = cast(Type[turu.core.async_cursor.GenericRowType], row_type)

    return cast(AsyncCursor, self)

executemany_with_tag(tag, operation, seq_of_parameters) async

Execute a database operation (Insert, Update, Delete) against all parameter sequences or mappings with a tag.

This is not defined in PEP 249,

This method executes an operation (Insert, Update, Delete) that does not return a value with a tag. This tag is used to verify that the specified operation is executed in order when testing with Mock.

Source code in turu-mysql/src/turu/mysql/async_cursor.py 
112
113
114
115
116
117
118
119
120
121
122
@override
async def executemany_with_tag(
    self,
    tag: Type[turu.core.tag.Tag],
    operation: str,
    seq_of_parameters: Sequence[Parameters],
) -> turu.core.async_cursor.AsyncCursor[Never, Parameters]:
    return cast(
        turu.core.async_cursor.AsyncCursor,
        await self.executemany(operation, seq_of_parameters),
    )

fetchall() async

Fetch all (remaining) rows of a query result

Source code in turu-mysql/src/turu/mysql/async_cursor.py 
145
146
147
148
149
@override
async def fetchall(self) -> List[turu.core.async_cursor.GenericRowType]:
    return [
        _map_row(self._row_type, row) for row in await self._raw_cursor.fetchall()
    ]

fetchmany(size=None) async

Fetch the next set of rows of a query result.

An empty sequence is returned when no more rows are available.

Parameters:

Name Type Description Default
size Optional[int]

The number of rows to fetch per call. If this parameter is not used, it is usually refer to use the .arraysize attribute (Default is 1). If this parameter is used, then it is best for it to retain the same value from one .fetchmany() call to the next.

 None
Source code in turu-mysql/src/turu/mysql/async_cursor.py 
132
133
134
135
136
137
138
139
140
141
142
143
@override
async def fetchmany(
    self, size: Optional[int] = None
) -> List[turu.core.async_cursor.GenericRowType]:
    return [
        _map_row(self._row_type, row)
        for row in (
            await self._raw_cursor.fetchmany(
                size if size is not None else self.arraysize
            )
        )
    ]

fetchone() async

Fetch the next row of a query result set.

Source code in turu-mysql/src/turu/mysql/async_cursor.py 
124
125
126
127
128
129
130
@override
async def fetchone(self) -> Optional[turu.core.async_cursor.GenericRowType]:
    row = await self._raw_cursor.fetchone()
    if row is None:
        return None

    return _map_row(self._row_type, row)