| 98 | |
| 99 | |
| 100 | class OraclePoolConnectionWrapper(PoolConnectionWrapper): |
| 101 | def _timestamp_convert(self, value: bytes) -> datetime.date: |
| 102 | try: |
| 103 | return parse_datetime(value.decode()).date() |
| 104 | except ValueError: |
| 105 | dt = parse_datetime(value.decode()[:-32]) |
| 106 | # Only convert to UTC-aware when use_tz=True |
| 107 | if get_use_tz(): |
| 108 | return dt.astimezone(tz=UTC) |
| 109 | else: |
| 110 | # When use_tz=False, strip timezone info to return naive datetime |
| 111 | return dt.replace(tzinfo=None) |
| 112 | |
| 113 | async def __aenter__(self) -> asyncodbc.Connection: |
| 114 | connection = await super().__aenter__() |
| 115 | if getattr(self.client, "database", False) and not hasattr(connection, "current_schema"): |
| 116 | client = cast(OracleClient, self.client) |
| 117 | await connection.execute(f'ALTER SESSION SET CURRENT_SCHEMA = "{client.user}"') |
| 118 | await connection.execute("ALTER SESSION SET NLS_DATE_FORMAT = 'YYYY-MM-DD'") |
| 119 | await connection.execute( |
| 120 | "ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT = 'YYYY-MM-DD\"T\"HH24:MI:SSTZH:TZM'" |
| 121 | ) |
| 122 | await connection.add_output_converter( |
| 123 | pyodbc.SQL_TYPE_TIMESTAMP, self._timestamp_convert |
| 124 | ) |
| 125 | connection.current_schema = client.user |
| 126 | return connection |
| 127 | |
| 128 | |
| 129 | class TransactionWrapper(ODBCTransactionWrapper, OracleClient): |
no outgoing calls
no test coverage detected
searching dependent graphs…