1
2
3
4
5
6
7
8
9
10
11
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
# Copyright (c) 2014, 2023, Oracle and/or its affiliates.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License, version 2.0, as
# published by the Free Software Foundation.
#
# This program is also distributed with certain software (including
# but not limited to OpenSSL) that is licensed under separate terms,
# as designated in a particular file or component or in included license
# documentation.  The authors of MySQL hereby grant you an
# additional permission to link the program and your derivative works
# with the separately licensed software that they have included with
# MySQL.
#
# Without limiting anything contained in the foregoing, this file,
# which is part of MySQL Connector/Python, is also subject to the
# Universal FOSS Exception, version 1.0, a copy of which can be found at
# http://oss.oracle.com/licenses/universal-foss-exception.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU General Public License, version 2.0, for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
 
# mypy: disable-error-code="arg-type,index"
 
"""Connection class using the C Extension."""
 
import os
import platform
import socket
 
from typing import Any, Dict, List, Optional, Sequence, Tuple, Type, Union
 
from . import version
from .abstracts import MySQLConnectionAbstract
from .constants import CharacterSet, ClientFlag, FieldFlag, ServerFlag, ShutdownType
from .conversion import MySQLConverter
from .errors import (
    InterfaceError,
    InternalError,
    OperationalError,
    ProgrammingError,
    get_mysql_exception,
)
from .protocol import MySQLProtocol
from .types import (
    CextEofPacketType,
    CextResultType,
    DescriptionType,
    ParamsSequenceOrDictType,
    RowType,
    StatsPacketType,
    StrOrBytes,
)
 
HAVE_CMYSQL = False
 
try:
    import _mysql_connector
 
    from _mysql_connector import MySQLInterfaceError, MySQLPrepStmt
 
    from .cursor_cext import (
        CMySQLCursor,
        CMySQLCursorBuffered,
        CMySQLCursorBufferedDict,
        CMySQLCursorBufferedNamedTuple,
        CMySQLCursorBufferedRaw,
        CMySQLCursorDict,
        CMySQLCursorNamedTuple,
        CMySQLCursorPrepared,
        CMySQLCursorPreparedDict,
        CMySQLCursorPreparedNamedTuple,
        CMySQLCursorPreparedRaw,
        CMySQLCursorRaw,
    )
 
    HAVE_CMYSQL = True
except ImportError as exc:
    raise ImportError(
        f"MySQL Connector/Python C Extension not available ({exc})"
    ) from exc
 
 
class CMySQLConnection(MySQLConnectionAbstract):
    """Class initiating a MySQL Connection using Connector/C."""
 
    def __init__(self, **kwargs: Any) -> None:
        """Initialization"""
        if not HAVE_CMYSQL:
            raise RuntimeError("MySQL Connector/Python C Extension not available")
        self._cmysql: Optional[
            _mysql_connector.MySQL  # pylint: disable=c-extension-no-member
        ] = None
        self._columns: List[DescriptionType] = []
        self._plugin_dir: str = os.path.join(
            os.path.dirname(os.path.abspath(_mysql_connector.__file__)),
            "mysql",
            "vendor",
            "plugin",
        )
        if platform.system() == "Linux":
            # Use the authentication plugins from system if they aren't bundled
            if not os.path.exists(self._plugin_dir):
                self._plugin_dir = (
                    "/usr/lib64/mysql/plugin"
                    if os.path.exists("/usr/lib64/mysql/plugin")
                    else "/usr/lib/mysql/plugin"
                )
 
        self.converter: Optional[MySQLConverter] = None
        super().__init__()
 
        if kwargs:
            self.connect(**kwargs)
 
    def _add_default_conn_attrs(self) -> None:
        """Add default connection attributes"""
        license_chunks = version.LICENSE.split(" ")
        if license_chunks[0] == "GPLv2":
            client_license = "GPL-2.0"
        else:
            client_license = "Commercial"
 
        self._conn_attrs.update(
            {
                "_connector_name": "mysql-connector-python",
                "_connector_license": client_license,
                "_connector_version": ".".join([str(x) for x in version.VERSION[0:3]]),
                "_source_host": socket.gethostname(),
            }
        )
 
    def _do_handshake(self) -> None:
        """Gather information of the MySQL server before authentication"""
        self._handshake = {
            "protocol": self._cmysql.get_proto_info(),
            "server_version_original": self._cmysql.get_server_info(),
            "server_threadid": self._cmysql.thread_id(),
            "charset": None,
            "server_status": None,
            "auth_plugin": None,
            "auth_data": None,
            "capabilities": self._cmysql.st_server_capabilities(),
        }
 
        self._server_version = self._check_server_version(
            self._handshake["server_version_original"]
        )
        CharacterSet.set_mysql_version(self._server_version)
 
    @property
    def _server_status(self) -> int:
        """Returns the server status attribute of MYSQL structure"""
        return self._cmysql.st_server_status()
 
    def set_allow_local_infile_in_path(self, path: str) -> None:
        """set local_infile_in_path
 
        Set allow_local_infile_in_path.
        """
 
        if self._cmysql:
            self._cmysql.set_load_data_local_infile_option(path)
 
    def set_unicode(self, value: bool = True) -> None:
        """Toggle unicode mode
 
        Set whether we return string fields as unicode or not.
        Default is True.
        """
        self._use_unicode = value
        if self._cmysql:
            self._cmysql.use_unicode(value)
        if self.converter:
            self.converter.set_unicode(value)
 
    @property
    def autocommit(self) -> bool:
        """Get whether autocommit is on or off"""
        value = self.info_query("SELECT @@session.autocommit")[0]
        return value == 1
 
    @autocommit.setter
    def autocommit(self, value: bool) -> None:
        """Toggle autocommit"""
        try:
            self._cmysql.autocommit(value)
            self._autocommit = value
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
    @property
    def database(self) -> str:
        """Get the current database"""
        return self.info_query("SELECT DATABASE()")[0]  # type: ignore[return-value]
 
    @database.setter
    def database(self, value: str) -> None:
        """Set the current database"""
        try:
            self._cmysql.select_db(value)
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
    @property
    def in_transaction(self) -> int:
        """MySQL session has started a transaction"""
        return self._server_status & ServerFlag.STATUS_IN_TRANS
 
    def _open_connection(self) -> None:
        charset_name = CharacterSet.get_info(self._charset_id)[0]
        # pylint: disable=c-extension-no-member
        self._cmysql = _mysql_connector.MySQL(
            buffered=self._buffered,
            raw=self._raw,
            charset_name=charset_name,
            connection_timeout=(self._connection_timeout or 0),
            use_unicode=self._use_unicode,
            auth_plugin=self._auth_plugin,
            plugin_dir=self._plugin_dir,
        )
        # pylint: enable=c-extension-no-member
        if not self.isset_client_flag(ClientFlag.CONNECT_ARGS):
            self._conn_attrs = {}
        cnx_kwargs = {
            "host": self._host,
            "user": self._user,
            "password": self._password,
            "password1": self._password1,
            "password2": self._password2,
            "password3": self._password3,
            "database": self._database,
            "port": self._port,
            "client_flags": self._client_flags,
            "unix_socket": self._unix_socket,
            "compress": self.isset_client_flag(ClientFlag.COMPRESS),
            "ssl_disabled": True,
            "conn_attrs": self._conn_attrs,
            "local_infile": self._allow_local_infile,
            "load_data_local_dir": self._allow_local_infile_in_path,
            "oci_config_file": self._oci_config_file,
            "oci_config_profile": self._oci_config_profile,
            "fido_callback": self._fido_callback,
        }
 
        tls_versions = self._ssl.get("tls_versions")
        if tls_versions is not None:
            tls_versions.sort(reverse=True)  # type: ignore[union-attr]
            tls_versions = ",".join(tls_versions)
        if self._ssl.get("tls_ciphersuites") is not None:
            ssl_ciphersuites = self._ssl.get("tls_ciphersuites")[0]
            tls_ciphersuites = self._ssl.get("tls_ciphersuites")[1]
        else:
            ssl_ciphersuites = None
            tls_ciphersuites = None
        if (
            tls_versions is not None
            and "TLSv1.3" in tls_versions
            and not tls_ciphersuites
        ):
            tls_ciphersuites = "TLS_AES_256_GCM_SHA384"
        if not self._ssl_disabled:
            cnx_kwargs.update(
                {
                    "ssl_ca": self._ssl.get("ca"),
                    "ssl_cert": self._ssl.get("cert"),
                    "ssl_key": self._ssl.get("key"),
                    "ssl_cipher_suites": ssl_ciphersuites,
                    "tls_versions": tls_versions,
                    "tls_cipher_suites": tls_ciphersuites,
                    "ssl_verify_cert": self._ssl.get("verify_cert") or False,
                    "ssl_verify_identity": self._ssl.get("verify_identity") or False,
                    "ssl_disabled": self._ssl_disabled,
                }
            )
 
        if os.name == "nt" and self._auth_plugin_class == "MySQLKerberosAuthPlugin":
            cnx_kwargs["use_kerberos_gssapi"] = True
 
        try:
            self._cmysql.connect(**cnx_kwargs)
            self._cmysql.converter_str_fallback = self._converter_str_fallback
            if self.converter:
                self.converter.str_fallback = self._converter_str_fallback
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
        self._do_handshake()
 
    def close(self) -> None:
        """Disconnect from the MySQL server"""
        if self._cmysql:
            try:
                self.free_result()
                self._cmysql.close()
            except MySQLInterfaceError as err:
                raise get_mysql_exception(
                    msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
                ) from err
 
    disconnect = close
 
    def is_closed(self) -> bool:
        """Return True if the connection to MySQL Server is closed."""
        return not self._cmysql.connected()
 
    def is_connected(self) -> bool:
        """Reports whether the connection to MySQL Server is available"""
        if self._cmysql:
            self.handle_unread_result()
            return self._cmysql.ping()
 
        return False
 
    def ping(self, reconnect: bool = False, attempts: int = 1, delay: int = 0) -> None:
        """Check availability of the MySQL server
 
        When reconnect is set to True, one or more attempts are made to try
        to reconnect to the MySQL server using the reconnect()-method.
 
        delay is the number of seconds to wait between each retry.
 
        When the connection is not available, an InterfaceError is raised. Use
        the is_connected()-method if you just want to check the connection
        without raising an error.
 
        Raises InterfaceError on errors.
        """
        self.handle_unread_result()
 
        try:
            connected = self._cmysql.ping()
        except AttributeError:
            pass  # Raise or reconnect later
        else:
            if connected:
                return
 
        if reconnect:
            self.reconnect(attempts=attempts, delay=delay)
        else:
            raise InterfaceError("Connection to MySQL is not available")
 
    def set_character_set_name(self, charset: str) -> None:
        """Sets the default character set name for current connection."""
        self._cmysql.set_character_set(charset)
 
    def info_query(self, query: StrOrBytes) -> Optional[RowType]:
        """Send a query which only returns 1 row"""
        first_row = ()
        try:
            self._cmysql.query(query)
            if self._cmysql.have_result_set:
                first_row = self._cmysql.fetch_row()
                if self._cmysql.fetch_row():
                    self._cmysql.free_result()
                    raise InterfaceError("Query should not return more than 1 row")
            self._cmysql.free_result()
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
        return first_row
 
    @property
    def connection_id(self) -> Optional[int]:
        """MySQL connection ID"""
        try:
            return self._cmysql.thread_id()
        except MySQLInterfaceError:
            pass  # Just return None
 
        return None
 
    def get_rows(
        self,
        count: Optional[int] = None,
        binary: bool = False,
        columns: Optional[List[DescriptionType]] = None,
        raw: Optional[bool] = None,
        prep_stmt: Optional[MySQLPrepStmt] = None,
    ) -> Tuple[List[RowType], Optional[CextEofPacketType]]:
        """Get all or a subset of rows returned by the MySQL server"""
        unread_result = prep_stmt.have_result_set if prep_stmt else self.unread_result
        if not (self._cmysql and unread_result):
            raise InternalError("No result set available")
 
        if raw is None:
            raw = self._raw
 
        rows: List[Tuple[Any, ...]] = []
        if count is not None and count <= 0:
            raise AttributeError("count should be 1 or higher, or None")
 
        counter = 0
        try:
            fetch_row = prep_stmt.fetch_row if prep_stmt else self._cmysql.fetch_row
            if self.converter:
                # When using a converter class, the C extension should not
                # convert the values. This can be accomplished by setting
                # the raw option to True.
                self._cmysql.raw(True)
            row = fetch_row()
            while row:
                if not self._raw and self.converter:
                    row = list(row)
                    for i, _ in enumerate(row):
                        if not raw:
                            row[i] = self.converter.to_python(self._columns[i], row[i])
                    row = tuple(row)
                rows.append(row)
                counter += 1
                if count and counter == count:
                    break
                row = fetch_row()
            if not row:
                _eof: Optional[CextEofPacketType] = self.fetch_eof_columns(prep_stmt)[
                    "eof"
                ]  # type: ignore[assignment]
                if prep_stmt:
                    prep_stmt.free_result()
                    self._unread_result = False
                else:
                    self.free_result()
            else:
                _eof = None
        except MySQLInterfaceError as err:
            if prep_stmt:
                prep_stmt.free_result()
                raise InterfaceError(str(err)) from err
            self.free_result()
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
        return rows, _eof
 
    def get_row(
        self,
        binary: bool = False,
        columns: Optional[List[DescriptionType]] = None,
        raw: Optional[bool] = None,
        prep_stmt: Optional[MySQLPrepStmt] = None,
    ) -> Tuple[Optional[RowType], CextEofPacketType]:
        """Get the next rows returned by the MySQL server"""
        try:
            rows, eof = self.get_rows(
                count=1,
                binary=binary,
                columns=columns,
                raw=raw,
                prep_stmt=prep_stmt,
            )
            if rows:
                return (rows[0], eof)
            return (None, eof)
        except IndexError:
            # No row available
            return (None, None)
 
    def next_result(self) -> Optional[bool]:
        """Reads the next result"""
        if self._cmysql:
            self._cmysql.consume_result()
            return self._cmysql.next_result()
        return None
 
    def free_result(self) -> None:
        """Frees the result"""
        if self._cmysql:
            self._cmysql.free_result()
 
    def commit(self) -> None:
        """Commit current transaction"""
        if self._cmysql:
            self.handle_unread_result()
            self._cmysql.commit()
 
    def rollback(self) -> None:
        """Rollback current transaction"""
        if self._cmysql:
            self._cmysql.consume_result()
            self._cmysql.rollback()
 
    def cmd_init_db(self, database: str) -> None:
        """Change the current database"""
        try:
            self._cmysql.select_db(database)
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
    def fetch_eof_columns(
        self, prep_stmt: Optional[MySQLPrepStmt] = None
    ) -> CextResultType:
        """Fetch EOF and column information"""
        have_result_set = (
            prep_stmt.have_result_set if prep_stmt else self._cmysql.have_result_set
        )
        if not have_result_set:
            raise InterfaceError("No result set")
 
        fields = prep_stmt.fetch_fields() if prep_stmt else self._cmysql.fetch_fields()
        self._columns = []
        for col in fields:
            self._columns.append(
                (
                    col[4],
                    int(col[8]),
                    None,
                    None,
                    None,
                    None,
                    ~int(col[9]) & FieldFlag.NOT_NULL,
                    int(col[9]),
                    int(col[6]),
                )
            )
 
        return {
            "eof": {
                "status_flag": self._server_status,
                "warning_count": self._cmysql.st_warning_count(),
            },
            "columns": self._columns,
        }
 
    def fetch_eof_status(self) -> Optional[CextEofPacketType]:
        """Fetch EOF and status information"""
        if self._cmysql:
            return {
                "warning_count": self._cmysql.st_warning_count(),
                "field_count": self._cmysql.st_field_count(),
                "insert_id": self._cmysql.insert_id(),
                "affected_rows": self._cmysql.affected_rows(),
                "server_status": self._server_status,
            }
 
        return None
 
    def cmd_stmt_prepare(self, statement: bytes) -> MySQLPrepStmt:
        """Prepares the SQL statement"""
        if not self._cmysql:
            raise OperationalError("MySQL Connection not available")
 
        try:
            stmt = self._cmysql.stmt_prepare(statement)
            stmt.converter_str_fallback = self._converter_str_fallback
            return stmt
        except MySQLInterfaceError as err:
            raise InterfaceError(str(err)) from err
 
    def cmd_stmt_execute(
        self, statement_id: MySQLPrepStmt, *args: Any
    ) -> Optional[Union[CextEofPacketType, CextResultType]]:
        """Executes the prepared statement"""
        try:
            statement_id.stmt_execute(*args)
        except MySQLInterfaceError as err:
            raise InterfaceError(str(err)) from err
 
        self._columns = []
        if not statement_id.have_result_set:
            # No result
            self._unread_result = False
            return self.fetch_eof_status()
 
        self._unread_result = True
        return self.fetch_eof_columns(statement_id)
 
    def cmd_stmt_close(self, statement_id: MySQLPrepStmt) -> None:
        """Closes the prepared statement"""
        if self._unread_result:
            raise InternalError("Unread result found")
        statement_id.stmt_close()
 
    def cmd_stmt_reset(self, statement_id: MySQLPrepStmt) -> None:
        """Resets the prepared statement"""
        if self._unread_result:
            raise InternalError("Unread result found")
        statement_id.stmt_reset()
 
    def cmd_query(
        self,
        query: StrOrBytes,
        raw: Optional[bool] = None,
        buffered: bool = False,
        raw_as_string: bool = False,
    ) -> Optional[Union[CextEofPacketType, CextResultType]]:
        """Send a query to the MySQL server"""
        self.handle_unread_result()
        if raw is None:
            raw = self._raw
        try:
            if not isinstance(query, bytes):
                query = query.encode("utf-8")
            self._cmysql.query(
                query,
                raw=raw,
                buffered=buffered,
                raw_as_string=raw_as_string,
                query_attrs=self._query_attrs,
            )
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                err.errno, msg=err.msg, sqlstate=err.sqlstate
            ) from err
        except AttributeError as err:
            addr = (
                self._unix_socket if self._unix_socket else f"{self._host}:{self._port}"
            )
            raise OperationalError(
                errno=2055, values=(addr, "Connection not available.")
            ) from err
 
        self._columns = []
        if not self._cmysql.have_result_set:
            # No result
            return self.fetch_eof_status()
 
        return self.fetch_eof_columns()
 
    _execute_query = cmd_query
 
    def cursor(
        self,
        buffered: Optional[bool] = None,
        raw: Optional[bool] = None,
        prepared: Optional[bool] = None,
        cursor_class: Optional[Type[CMySQLCursor]] = None,
        dictionary: Optional[bool] = None,
        named_tuple: Optional[bool] = None,
    ) -> CMySQLCursor:
        """Instantiates and returns a cursor using C Extension
 
        By default, CMySQLCursor is returned. Depending on the options
        while connecting, a buffered and/or raw cursor is instantiated
        instead. Also depending upon the cursor options, rows can be
        returned as dictionary or named tuple.
 
        Dictionary and namedtuple based cursors are available with buffered
        output but not raw.
 
        It is possible to also give a custom cursor through the
        cursor_class parameter, but it needs to be a subclass of
        mysql.connector.cursor_cext.CMySQLCursor.
 
        Raises ProgrammingError when cursor_class is not a subclass of
        CursorBase. Raises ValueError when cursor is not available.
 
        Returns instance of CMySQLCursor or subclass.
 
        :param buffered: Return a buffering cursor
        :param raw: Return a raw cursor
        :param prepared: Return a cursor which uses prepared statements
        :param cursor_class: Use a custom cursor class
        :param dictionary: Rows are returned as dictionary
        :param named_tuple: Rows are returned as named tuple
        :return: Subclass of CMySQLCursor
        :rtype: CMySQLCursor or subclass
        """
        self.handle_unread_result(prepared)
        if not self.is_connected():
            raise OperationalError("MySQL Connection not available.")
        if cursor_class is not None:
            if not issubclass(cursor_class, CMySQLCursor):
                raise ProgrammingError(
                    "Cursor class needs be to subclass of cursor_cext.CMySQLCursor"
                )
            return (cursor_class)(self)
 
        buffered = buffered or self._buffered
        raw = raw or self._raw
 
        cursor_type = 0
        if buffered is True:
            cursor_type |= 1
        if raw is True:
            cursor_type |= 2
        if dictionary is True:
            cursor_type |= 4
        if named_tuple is True:
            cursor_type |= 8
        if prepared is True:
            cursor_type |= 16
 
        types = {
            0: CMySQLCursor,  # 0
            1: CMySQLCursorBuffered,
            2: CMySQLCursorRaw,
            3: CMySQLCursorBufferedRaw,
            4: CMySQLCursorDict,
            5: CMySQLCursorBufferedDict,
            8: CMySQLCursorNamedTuple,
            9: CMySQLCursorBufferedNamedTuple,
            16: CMySQLCursorPrepared,
            18: CMySQLCursorPreparedRaw,
            20: CMySQLCursorPreparedDict,
            24: CMySQLCursorPreparedNamedTuple,
        }
        try:
            return (types[cursor_type])(self)
        except KeyError:
            args = ("buffered", "raw", "dictionary", "named_tuple", "prepared")
            raise ValueError(
                "Cursor not available with given criteria: "
                + ", ".join([args[i] for i in range(5) if cursor_type & (1 << i) != 0])
            ) from None
 
    @property
    def num_rows(self) -> int:
        """Returns number of rows of current result set"""
        if not self._cmysql.have_result_set:
            raise InterfaceError("No result set")
 
        return self._cmysql.num_rows()
 
    @property
    def warning_count(self) -> int:
        """Returns number of warnings"""
        if not self._cmysql:
            return 0
 
        return self._cmysql.warning_count()
 
    @property
    def result_set_available(self) -> bool:
        """Check if a result set is available"""
        if not self._cmysql:
            return False
 
        return self._cmysql.have_result_set
 
    @property  # type: ignore[misc]
    def unread_result(self) -> bool:
        """Check if there are unread results or rows"""
        return self.result_set_available
 
    @property
    def more_results(self) -> bool:
        """Check if there are more results"""
        return self._cmysql.more_results()
 
    def prepare_for_mysql(
        self, params: ParamsSequenceOrDictType
    ) -> Union[Sequence[bytes], Dict[str, bytes],]:
        """Prepare parameters for statements
 
        This method is use by cursors to prepared parameters found in the
        list (or tuple) params.
 
        Returns dict.
        """
        result: Union[List[Any], Dict[str, Any]] = []
        if isinstance(params, (list, tuple)):
            if self.converter:
                result = [
                    self.converter.quote(
                        self.converter.escape(
                            self.converter.to_mysql(value), self._sql_mode
                        )
                    )
                    for value in params
                ]
            else:
                result = self._cmysql.convert_to_mysql(*params)
        elif isinstance(params, dict):
            result = {}
            if self.converter:
                for key, value in params.items():
                    result[key] = self.converter.quote(
                        self.converter.escape(
                            self.converter.to_mysql(value), self._sql_mode
                        )
                    )
            else:
                for key, value in params.items():
                    result[key] = self._cmysql.convert_to_mysql(value)[0]
        else:
            raise ProgrammingError(
                f"Could not process parameters: {type(params).__name__}({params}),"
                " it must be of type list, tuple or dict"
            )
 
        return result
 
    def consume_results(self) -> None:
        """Consume the current result
 
        This method consume the result by reading (consuming) all rows.
        """
        self._cmysql.consume_result()
 
    def cmd_change_user(
        self,
        username: str = "",
        password: str = "",
        database: str = "",
        charset: int = 45,
        password1: str = "",
        password2: str = "",
        password3: str = "",
        oci_config_file: Optional[str] = None,
        oci_config_profile: Optional[str] = None,
    ) -> None:
        """Change the current logged in user"""
        try:
            self._cmysql.change_user(
                username,
                password,
                database,
                password1,
                password2,
                password3,
                oci_config_file,
                oci_config_profile,
            )
 
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
        self._charset_id = charset
        self._post_connection()
 
    def cmd_reset_connection(self) -> bool:
        """Resets the session state without re-authenticating
 
        Reset command only works on MySQL server 5.7.3 or later.
        The result is True for a successful reset otherwise False.
 
        Returns bool
        """
        res = self._cmysql.reset_connection()
        if res:
            self._post_connection()
        return res
 
    def cmd_refresh(self, options: int) -> Optional[CextEofPacketType]:
        """Send the Refresh command to the MySQL server"""
        try:
            self.handle_unread_result()
            self._cmysql.refresh(options)
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
        return self.fetch_eof_status()
 
    def cmd_quit(self) -> None:
        """Close the current connection with the server"""
        self.close()
 
    def cmd_shutdown(self, shutdown_type: Optional[int] = None) -> None:
        """Shut down the MySQL Server"""
        if not self._cmysql:
            raise OperationalError("MySQL Connection not available")
 
        if shutdown_type:
            if not ShutdownType.get_info(shutdown_type):
                raise InterfaceError("Invalid shutdown type")
            level = shutdown_type
        else:
            level = ShutdownType.SHUTDOWN_DEFAULT
 
        try:
            self._cmysql.shutdown(level)
        except MySQLInterfaceError as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
        self.close()
 
    def cmd_statistics(self) -> StatsPacketType:
        """Return statistics from the MySQL server"""
        self.handle_unread_result()
 
        try:
            stat = self._cmysql.stat()
            return MySQLProtocol().parse_statistics(stat, with_header=False)
        except (MySQLInterfaceError, InterfaceError) as err:
            raise get_mysql_exception(
                msg=err.msg, errno=err.errno, sqlstate=err.sqlstate
            ) from err
 
    def cmd_process_kill(self, mysql_pid: int) -> None:
        """Kill a MySQL process"""
        if not isinstance(mysql_pid, int):
            raise ValueError("MySQL PID must be int")
        self.info_query(f"KILL {mysql_pid}")
 
    def cmd_debug(self) -> Any:
        """Send the DEBUG command"""
        raise NotImplementedError
 
    def cmd_ping(self) -> Any:
        """Send the PING command"""
        raise NotImplementedError
 
    def cmd_query_iter(self, statements: Any) -> Any:
        """Send one or more statements to the MySQL server"""
        raise NotImplementedError
 
    def cmd_stmt_send_long_data(
        self, statement_id: Any, param_id: Any, data: Any
    ) -> Any:
        """Send data for a column"""
        raise NotImplementedError
 
    def handle_unread_result(self, prepared: bool = False) -> None:
        """Check whether there is an unread result"""
        unread_result = self._unread_result if prepared is True else self.unread_result
        if self.can_consume_results:
            self.consume_results()
        elif unread_result:
            raise InternalError("Unread result found")
 
    def reset_session(
        self,
        user_variables: Optional[Dict[str, Any]] = None,
        session_variables: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Clears the current active session
 
        This method resets the session state, if the MySQL server is 5.7.3
        or later active session will be reset without re-authenticating.
        For other server versions session will be reset by re-authenticating.
 
        It is possible to provide a sequence of variables and their values to
        be set after clearing the session. This is possible for both user
        defined variables and session variables.
        This method takes two arguments user_variables and session_variables
        which are dictionaries.
 
        Raises OperationalError if not connected, InternalError if there are
        unread results and InterfaceError on errors.
        """
        if not self.is_connected():
            raise OperationalError("MySQL Connection not available.")
 
        if not self.cmd_reset_connection():
            try:
                self.cmd_change_user(
                    self._user,
                    self._password,
                    self._database,
                    self._charset_id,
                    self._password1,
                    self._password2,
                    self._password3,
                    self._oci_config_file,
                    self._oci_config_profile,
                )
            except ProgrammingError:
                self.reconnect()
 
        if user_variables or session_variables:
            cur = self.cursor()
            if user_variables:
                for key, value in user_variables.items():
                    cur.execute(f"SET @`{key}` = %s", (value,))
            if session_variables:
                for key, value in session_variables.items():
                    cur.execute(f"SET SESSION `{key}` = %s", (value,))
            cur.close()