U Dx`E;@sUdZddlZddlZddlmZmZmZmZmZm Z m Z ddl Z ddl Z ddlZddlZddlmZddlmZddlmZddlmZddlmZeeZejed <ejj dd d d Z!eee"e"dd ddZ#eee"e"e$d ddZ%e j&ee"e"e"e$eee"e"feee"e'fdd ddZ(d'ee"ee"ee"ee"ee j)ee'ee'e'ejj d ddZ*d(e"ejj ee e"ee"fee eee edfeeeffee'eee"ej+fe$e e j&ee j&fdddZ,d)e"ejj ee"ee e"ee"fee eee edfeeeffee'eee"ej+fe$e e j&ee j&fd dd Z-ed*e j&ejj e"e"e"e$eee"e"feee"e'fe$e'dd$ d%d&Z.dS)+zAmazon MySQL Module.N)AnyDictIteratorListOptionalTupleUnion)Cursor) _data_types) _databases) exceptions) apply_configs_logger)conreturncCst|tjjstddS)NzInvalid 'conn' argument, please pass a pymysql.connections.Connection object. Use pymysql.connect() to use credentials directly or wr.mysql.connect() to fetch it from the Glue Catalog.) isinstancepymysql connections Connectionr ZInvalidConnectionrr8/tmp/pip-target-zr53vnty/lib/python/awswrangler/mysql.py_validate_connectionsr)cursorschematablercCs@|rd|dnd}d|d|d}td|||dS)N`z`.zDROP TABLE IF EXISTS zDrop table query: %s)rdebugexecute)rrr schema_strsqlrrr _drop_tables r"cCs<|rd|dnd}|d|d|dt|dkS)NzTABLE_SCHEMA = 'z' ANDrz.SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE z TABLE_NAME = ''r)rlenZfetchall)rrrr rrr_does_table_exist%sr%) dfrrrmodeindexdtypevarchar_lengthsrc Cs|dkrt|||dnt|||dr*dStj|||d|tjd}ddd|Ddd} d |d |d | d } td | | | dS)N overwrite)rrrZTEXT)r&r(r)Zvarchar_lengths_defaultr*Zconverter_funcrcSs"g|]\}}d|d|dqS)r` z, r).0kvrrr Asz!_create_table..zCREATE TABLE IF NOT EXISTS ``.`z` ( )zCreate table query: %s) r"r%r Zdatabase_types_from_pandasZ pyarrow2mysqljoinitemsrrr) r&rrrr'r(r)r*Z mysql_typesZcols_strr!rrr _create_table+s   r6 ) connection secret_id catalog_iddbname boto3_session read_timeout write_timeoutconnect_timeoutrc CsVtj|||||d}|jdkr2td|jdtj|j|j|j |j |j |||dS)a(Return a pymysql connection from a Glue Catalog Connection. https://pymysql.readthedocs.io Parameters ---------- connection : str Glue Catalog Connection name. secret_id: Optional[str]: Specifies the secret containing the version that you want to retrieve. You can specify either the Amazon Resource Name (ARN) or the friendly name of the secret. catalog_id : str, optional The ID of the Data Catalog. If none is provided, the AWS account ID is used by default. dbname: Optional[str] Optional database name to overwrite the stored one. boto3_session : boto3.Session(), optional Boto3 Session. The default boto3 session will be used if boto3_session receive None. read_timeout: Optional[int] The timeout for reading from the connection in seconds (default: None - no timeout). This parameter is forward to pymysql. https://pymysql.readthedocs.io/en/latest/modules/connections.html write_timeout: Optional[int] The timeout for writing to the connection in seconds (default: None - no timeout) This parameter is forward to pymysql. https://pymysql.readthedocs.io/en/latest/modules/connections.html connect_timeout: int Timeout before throwing an exception when connecting. (default: 10, min: 1, max: 31536000) This parameter is forward to pymysql. https://pymysql.readthedocs.io/en/latest/modules/connections.html Returns ------- pymysql.connections.Connection pymysql connection. Examples -------- >>> import awswrangler as wr >>> con = wr.mysql.connect("MY_GLUE_CONNECTION") >>> with con.cursor() as cursor: >>> cursor.execute("SELECT 1") >>> print(cursor.fetchall()) >>> con.close() )r8r9r:r;r<ZmysqlzInvalid connection type (z!. It must be a MySQL connection.))userdatabasepasswordporthostr=r>r?) _db_utilsZget_connection_attributeskindr ZInvalidDatabaseTyperconnectr@rArBrCrD) r8r9r:r;r<r=r>r?attrsrrrrGGs&9 rGT.)r!r index_colparams chunksizer)saferc Cs"t|dtj|||||||dS)uReturn a DataFrame corresponding to the result set of the query string. Parameters ---------- sql : str SQL query. con : pymysql.connections.Connection Use pymysql.connect() to use credentials directly or wr.mysql.connect() to fetch it from the Glue Catalog. index_col : Union[str, List[str]], optional Column(s) to set as index(MultiIndex). params : Union[List, Tuple, Dict], optional List of parameters to pass to execute method. The syntax used to pass parameters is database driver dependent. Check your database driver documentation for which of the five syntax styles, described in PEP 249’s paramstyle, is supported. chunksize : int, optional If specified, return an iterator where chunksize is the number of rows to include in each chunk. dtype : Dict[str, pyarrow.DataType], optional Specifying the datatype for columns. The keys should be the column names and the values should be the PyArrow types. safe : bool Check for overflows or other unsafe data type conversions. Returns ------- Union[pandas.DataFrame, Iterator[pandas.DataFrame]] Result as Pandas DataFrame(s). Examples -------- Reading from MySQL using a Glue Catalog Connections >>> import awswrangler as wr >>> con = wr.mysql.connect("MY_GLUE_CONNECTION") >>> df = wr.mysql.read_sql_query( ... sql="SELECT * FROM test.my_table", ... con=con ... ) >>> con.close() rr!rrIrJrKr)rL)rrEread_sql_queryrMrrrrNs2 rN) rrrrIrJrKr)rLrc Cs<|dkrd|dnd|d|d}t|||||||dS)ueReturn a DataFrame corresponding the table. Parameters ---------- table : str Table name. con : pymysql.connections.Connection Use pymysql.connect() to use credentials directly or wr.mysql.connect() to fetch it from the Glue Catalog. schema : str, optional Name of SQL schema in database to query. Uses default schema if None. index_col : Union[str, List[str]], optional Column(s) to set as index(MultiIndex). params : Union[List, Tuple, Dict], optional List of parameters to pass to execute method. The syntax used to pass parameters is database driver dependent. Check your database driver documentation for which of the five syntax styles, described in PEP 249’s paramstyle, is supported. chunksize : int, optional If specified, return an iterator where chunksize is the number of rows to include in each chunk. dtype : Dict[str, pyarrow.DataType], optional Specifying the datatype for columns. The keys should be the column names and the values should be the PyArrow types. safe : bool Check for overflows or other unsafe data type conversions. Returns ------- Union[pandas.DataFrame, Iterator[pandas.DataFrame]] Result as Pandas DataFrame(s). Examples -------- Reading from MySQL using a Glue Catalog Connections >>> import awswrangler as wr >>> con = wr.mysql.connect("MY_GLUE_CONNECTION") >>> df = wr.mysql.read_sql_table( ... table="my_table", ... schema="test", ... con=con ... ) >>> con.close() NzSELECT * FROM `rr2rM)rN) rrrrIrJrKr)rLr!rrrread_sql_tables7&rOappendF) r&rrrr'r(r)r*use_column_namesrKrc  Csl|jdkrt|}dddddg} || krJtdd| t|d z|} t || ||||||d |r|j |j j dd dd gt |j} d } d }d }|rdd|jd} |dkrd|jdd}d|}tj|| | d}|D]l\}}|dkr8d|d|d| d|}n d|d|d| d|| }td|| ||fq||dkr"|dtj}| d|d|d|d|d | d|d|d|d|d | d|d|d| d |d|d!|d|W5QRXWn8tk rf}z|t|W5d"}~XYnXd"S)#a Write records stored in a DataFrame into MySQL. Parameters ---------- df : pandas.DataFrame Pandas DataFrame https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html con : pymysql.connections.Connection Use pymysql.connect() to use credentials directly or wr.mysql.connect() to fetch it from the Glue Catalog. table : str Table name schema : str Schema name mode : str Append, overwrite, upsert_duplicate_key, upsert_replace_into, upsert_distinct. append: Inserts new records into table overwrite: Drops table and recreates upsert_duplicate_key: Performs an upsert using `ON DUPLICATE KEY` clause. Requires table schema to have defined keys, otherwise duplicate records will be inserted. upsert_replace_into: Performs upsert using `REPLACE INTO` clause. Less efficient and still requires the table schema to have keys or else duplicate records will be inserted upsert_distinct: Inserts new records, including duplicates, then recreates the table and inserts `DISTINCT` records from old table. This is the least efficient approach but handles scenarios where there are no keys on table. index : bool True to store the DataFrame index as a column in the table, otherwise False to ignore it. dtype: Dict[str, str], optional Dictionary of columns names and MySQL types to be casted. Useful when you have columns with undetermined or mixed data types. (e.g. {'col name': 'TEXT', 'col2 name': 'FLOAT'}) varchar_lengths : Dict[str, int], optional Dict of VARCHAR length by columns. (e.g. {"col1": 10, "col5": 200}). use_column_names: bool If set to True, will use the column names of the DataFrame for generating the INSERT SQL Query. E.g. If the DataFrame has two columns `col1` and `col3` and `use_column_names` is True, data will only be inserted into the database columns `col1` and `col3`. chunksize: int Number of rows which are inserted with each SQL query. Defaults to inserting 200 rows per query. Returns ------- None None. Examples -------- Writing to MySQL using a Glue Catalog Connections >>> import awswrangler as wr >>> con = wr.mysql.connect("MY_GLUE_CONNECTION") >>> wr.mysql.to_sql( ... df=df, ... table="my_table", ... schema="test", ... con=con ... ) >>> con.close() TrPr+Zupsert_replace_intoZupsert_duplicate_keyZupsert_distinctzmode must be one of z, r)r&rrrr'r(r)r*)levelZinplacez%sr(r3cSsd|d|dS)Nrz `=VALUES(`z`)r)columnrrrrzto_sql..z ON DUPLICATE KEY UPDATE )r&column_placeholdersrKzREPLACE INTO `r2r,z VALUES z INSERT INTO `zsql: %s_zCREATE TABLE `z` LIKE `rz` SELECT DISTINCT * FROM `zDROP TABLE IF EXISTS `z ALTER TABLE `z ` RENAME TO `N)emptyr ZEmptyDataFramestriplowerZInvalidArgumentValuer4rrr6Z reset_indexr(namesr$columnsmaprEZ$generate_placeholder_parameter_pairsrrZ executemanycommituuiduuid4hexr Exceptionrollbackerror)r&rrrr'r(r)r*rRrKmodesrrXZinsertion_columnsZupsert_columnsZ upsert_strZ$placeholder_parameter_pair_generatorZ placeholders parametersr!Z temp_tableexrrrto_sqlstI           $$ rj)NNNNNNNr7)NNNNT)NNNNNT)rPFNNFrQ)/__doc__loggingratypingrrrrrrrZboto3ZpandaspdZpyarrowparZpymysql.cursorsr Z awswranglerr r rEr Zawswrangler._configr getLogger__name__rLogger__annotations__rrrstrr"boolr%Z DataFrameintr6SessionrGZDataTyperNrOrjrrrrs$        M$ ;$ =