

import sys
from dataclasses import dataclass
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union

import pandas as pd
import pyarrow as pa

import datasets
import datasets.config
from datasets.features.features import require_storage_cast
from datasets.table import table_cast


if TYPE_CHECKING:
 import sqlitea

 import sqlalchemy


_UpperCamelCase			      =		datasets.utils.logging.get_logger(__name__)







@dataclass
class   __lowercase				(datasets.BuilderConfig			):
 _UpperCamelCase        =  None
 _UpperCamelCase        =  None
 _UpperCamelCase        =  None
 _UpperCamelCase        =  True
 _UpperCamelCase        =  None
 _UpperCamelCase        =  None
 _UpperCamelCase        =  None
 _UpperCamelCase        =  10000
 _UpperCamelCase        =  None
 def      UpperCamelCase__  (							self ) ->List[str]:



       '''simple docstring'''





       if self.sql is None:
             raise ValueError('''sql must be specified''' )
       if self.con is None:
             raise ValueError('''con must be specified''' )
 def      UpperCamelCase__  (							self				,							A_				,							A_ = None				,							) ->str:



       '''simple docstring'''





       __lowerCAmelCase :						str = config_kwargs.copy()
       # We need to stringify the Selectable object to make its hash deterministic

       # The process of stringifying is explained here: http://docs.sqlalchemy.org/en/latest/faq/sqlexpressions.html
       __lowerCAmelCase :						Optional[int] = config_kwargs['''sql''']
       if not isinstance(A_				,							A_ ):
             if datasets.config.SQLALCHEMY_AVAILABLE and "sqlalchemy" in sys.modules:
                   import sqlalchemy

                   if isinstance(A_				,							sqlalchemy.sql.Selectable ):
                         __lowerCAmelCase :						Optional[Any] = sqlalchemy.create_engine(config_kwargs['''con'''].split('''://''' )[0] + '''://''' )
                         __lowerCAmelCase :						List[Any] = str(sql.compile(dialect=engine.dialect ) )
                         __lowerCAmelCase :						List[Any] = sql_str
                   else:
                         raise TypeError(
                             f"""Supported types for 'sql' are string and sqlalchemy.sql.Selectable but got {type(A_ )}: {sql}""" )
             else:
                   raise TypeError(
                       f"""Supported types for 'sql' are string and sqlalchemy.sql.Selectable but got {type(A_ )}: {sql}""" )
       __lowerCAmelCase :						Any = config_kwargs['''con''']
       if not isinstance(A_				,							A_ ):
             __lowerCAmelCase :						List[Any] = id(A_ )
             logger.info(
                 f"""SQL connection 'con' of type {type(A_ )} couldn't be hashed properly. To enable hashing, specify 'con' as URI string instead.""" )

       return super().create_config_id(A_				,							custom_features=A_ )





 @property
 def      UpperCamelCase__  (							self ) ->List[Any]:



       '''simple docstring'''





       __lowerCAmelCase :						Any = {
           '''index_col''': self.index_col,
           '''columns''': self.columns,
           '''params''': self.params,
           '''coerce_float''': self.coerce_float,
           '''parse_dates''': self.parse_dates,
       }
       return pd_read_sql_kwargs







class   __lowercase				(datasets.ArrowBasedBuilder			):
 _UpperCamelCase        =  SqlConfig
 def      UpperCamelCase__  (							self ) ->List[Any]:



       '''simple docstring'''





       return datasets.DatasetInfo(features=self.config.features )
 def      UpperCamelCase__  (							self				,							A_ ) ->List[Any]:



       '''simple docstring'''





       return [datasets.SplitGenerator(name=datasets.Split.TRAIN				,							gen_kwargs={} )]
 def      UpperCamelCase__  (							self				,							A_ ) ->pa.Table:



       '''simple docstring'''





       if self.config.features is not None:
             __lowerCAmelCase :						str = self.config.features.arrow_schema
             if all(not require_storage_cast(A_ ) for feature in self.config.features.values() ):
                   # cheaper cast
                   __lowerCAmelCase :						Tuple = pa.Table.from_arrays([pa_table[field.name] for field in schema]				,							schema=A_ )
             else:
                   # more expensive cast; allows str <-> int/float or str to Audio for example
                   __lowerCAmelCase :						Union[str, Any] = table_cast(A_				,							A_ )
       return pa_table





 def      UpperCamelCase__  (							self ) ->Tuple:



       '''simple docstring'''





       __lowerCAmelCase :						Optional[int] = self.config.chunksize
       __lowerCAmelCase :						Optional[int] = pd.read_sql(
           self.config.sql				,							self.config.con				,							chunksize=A_				,							**self.config.pd_read_sql_kwargs )
       __lowerCAmelCase :						int = [sql_reader] if chunksize is None else sql_reader
       for chunk_idx, df in enumerate(A_ ):
             __lowerCAmelCase :						Any = pa.Table.from_pandas(A_ )
             yield chunk_idx, self._cast_table(A_ )





def 		_lowercase			(					lowercase__						):
      return "".join(chr(ord(lowercase__						) + 3_2						) if '''A''' <= char <= '''Z''' else char for char in word						)


if __name__ == "__main__":
 from doctest import testmod

 testmod()


