
    Bh                    F   S r SSKJr  SSKrSSKrSSKJr  SSKJr  SSKJr  SSKJ	r	  SSKJ
r
  SS	KJr  S
SKJr  S
SKJr  S
SKJr  SSKJr  SSKJr  SSKJr  SSKJr  \
(       a  SSKJr  SSKJr  SSKJr  SSKJr   " S S\5      r " S S\5      r  " S S\\5      r!\!r"g)aH  

.. dialect:: mysql+pyodbc
    :name: PyODBC
    :dbapi: pyodbc
    :connectstring: mysql+pyodbc://<username>:<password>@<dsnname>
    :url: https://pypi.org/project/pyodbc/

.. note::

    The PyODBC for MySQL dialect is **not tested as part of
    SQLAlchemy's continuous integration**.
    The recommended MySQL dialects are mysqlclient and PyMySQL.
    However, if you want to use the mysql+pyodbc dialect and require
    full support for ``utf8mb4`` characters (including supplementary
    characters like emoji) be sure to use a current release of
    MySQL Connector/ODBC and specify the "ANSI" (**not** "Unicode")
    version of the driver in your DSN or connection string.

Pass through exact pyodbc connection string::

    import urllib

    connection_string = (
        "DRIVER=MySQL ODBC 8.0 ANSI Driver;"
        "SERVER=localhost;"
        "PORT=3307;"
        "DATABASE=mydb;"
        "UID=root;"
        "PWD=(whatever);"
        "charset=utf8mb4;"
    )
    params = urllib.parse.quote_plus(connection_string)
    connection_uri = "mysql+pyodbc:///?odbc_connect=%s" % params

    )annotationsN)Any)Callable)Optional)Tuple)TYPE_CHECKING)Union   )MySQLDialect)MySQLExecutionContext)TIME   )exc)util)PyODBCConnector)Time)
Connection)DBAPIConnection)Dialect)_ResultProcessorTypec                  *    \ rS rSr      SS jrSrg)_pyodbcTIMEG   c                    SS jnU$ )Nc                    U $ N )values    c/home/kali/devsecops-assessor/venv/lib/python3.13/site-packages/sqlalchemy/dialects/mysql/pyodbc.pyprocess-_pyodbcTIME.result_processor.<locals>.processK   s    L    )r   r   returnzUnion[datetime.time, None]r   )selfdialectcoltyper    s       r   result_processor_pyodbcTIME.result_processorH   s    	 r"   r   N)r%   r   r&   objectr#   z#_ResultProcessorType[datetime.time])__name__
__module____qualname____firstlineno__r'   __static_attributes__r   r"   r   r   r   G   s    )/	,r"   r   c                      \ rS rSrSS jrSrg)MySQLExecutionContext_pyodbcR   c                    U R                  5       nUR                  S5        UR                  5       S   nUR                  5         U$ )NzSELECT LAST_INSERT_ID()r   )create_cursorexecutefetchoneclose)r$   cursor	lastrowids      r   get_lastrowid*MySQLExecutionContext_pyodbc.get_lastrowidS   s=    ##%01OO%a(	r"   r   N)r#   int)r*   r+   r,   r-   r9   r.   r   r"   r   r0   r0   R   s    r"   r0   c                     ^  \ rS rSrSr\R                  " \R                  \	\
05      rSr\rSrS	S jr    S
S jrSS jrSU 4S jjrSrU =r$ )MySQLDialect_pyodbc[   TMySQLc                    SU l          U R                  US5      nU(       a  U$  [        R
                  " S5        g! [        R                   a     N-f = f)z:Sniff out the character set in use for connection results.Ncharacter_set_clientz@Could not detect the connection character set.  Assuming latin1.latin1)_connection_charset_fetch_settingr   
DBAPIErrorr   warn)r$   
connectionr   s      r   _detect_charset#MySQLDialect_pyodbc._detect_charsetc   sa     $( 	''
4JKE 
 				
  ~~ 		s   < AAc                .    [         R                  " X5      $ r   )r   _get_server_version_info)r$   rG   s     r   rK   ,MySQLDialect_pyodbc._get_server_version_info|   s     44TFFr"   c                    [         R                  " S5      R                  [        UR                  5      5      nUc  g UR                  S5      nU(       a  [        U5      $ g )Nz	\((\d+)\)r
   )recompilesearchstrargsgroupr;   )r$   	exceptionmcs       r   _extract_error_code'MySQLDialect_pyodbc._extract_error_code   sH    JJ|$++C	,?@9771:q6Mr"   c                6   >^ [         TU ]  5       mSU4S jjnU$ )Nc                   > Tb  T" U 5        SnSnU R                  USS9  U R                  USS9  U R                  SS9  g )Nr
   izutf-8)encoding)setdecodingsetencoding)connpyodbc_SQL_CHARpyodbc_SQL_WCHARsuper_s      r   
on_connect2MySQLDialect_pyodbc.on_connect.<locals>.on_connect   sU    !t  O!_w?-@g.r"   )r^   r   r#   None)superrb   )r$   rb   ra   	__class__s     @r   rb   MySQLDialect_pyodbc.on_connect   s    #%
	/ r"   )rC   )rG   r   r#   rQ   )rG   r   r#   zTuple[int, ...])rT   BaseExceptionr#   zOptional[int])r#   z!Callable[[DBAPIConnection], None])r*   r+   r,   r-   supports_statement_cacher   update_copyr   colspecsr   r   supports_unicode_statementsr0   execution_ctx_clspyodbc_driver_namerH   rK   rW   rb   r.   __classcell__)rf   s   @r   r=   r=   [   sd    # 5 5k7JKH"&4 2G$G	G
 r"   r=   )#__doc__
__future__r   datetimerN   typingr   r   r   r   r   r	   baser   r   typesr    r   r   connectors.pyodbcr   sql.sqltypesr   enginer   engine.interfacesr   r   sql.type_apir   r   r0   r=   r%   r   r"   r   <module>r|      s{   #H #  	         '    0  $4,4$ #8 ?/< ?D r"   