U Dx`3@sUdZddlZddlmZddlmZmZmZmZm Z m Z m Z ddl Z ddl ZddlZddlZddlmZddlmZddlmZddlmZeeZejed <ejdd d d Zej e e!e!dd ddZ"ej e e!e!e#d ddZ$ej%ej e!e!e!e#e ee!e!fe ee!e&fdd ddZ'd&e e!e e!e e!e e!e e j(e e e#efe e&e#ejd ddZ)d'e!eje e e!ee!fe e eee edfeeeffe e&e ee!ej*fe#e ej%eej%fdddZ+d(e!eje e!e e e!ee!fe e eee edfeeeffe e&e ee!ej*fe#e ej%eej%fd ddZ,ed)ej%eje!e!e!e#e ee!e!fe ee!e&fe#e&dd# d$d%Z-dS)*zAmazon PostgreSQL Module.N) SSLContext)AnyDictIteratorListOptionalTupleUnion) _data_types) _databases) exceptions) apply_configs_logger)conreturncCst|tjstddS)NzInvalid 'conn' argument, please pass a pg8000.Connection object. Use pg8000.connect() to use credentials directly or wr.postgresql.connect() to fetch it from the Glue Catalog.) isinstancepg8000 Connectionr ZInvalidConnectionrr=/tmp/pip-target-zr53vnty/lib/python/awswrangler/postgresql.py_validate_connections r)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' ANDrzHSELECT true WHERE EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE z TABLE_NAME = 'z');r)rlenZfetchall)rrrrrrr_does_table_exist$s r#) 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|dqS) z, r).0kvrrr Esz!_create_table..zCREATE TABLE IF NOT EXISTS ""."z" ( )zCreate table query: %s) r!r#r Zdatabase_types_from_pandasZpyarrow2postgresqljoinitemsrrr) r$rrrr%r&r'r(Zpostgresql_typesZcols_strr rrr _create_table/s   r4T) connection secret_id catalog_iddbname boto3_session ssl_contexttimeout tcp_keepaliverc Cs`tj|||||d}|jdkr<|jdkr>> import awswrangler as wr >>> con = wr.postgresql.connect("MY_GLUE_CONNECTION") >>> with con.cursor() as cursor: >>> cursor.execute("SELECT 1") >>> print(cursor.fetchall()) >>> con.close() )r5r6r7r8r9Z postgresqlZpostgreszInvalid connection type (z&. It must be a postgresql connection.))userdatabasepasswordporthostr:r;r<) _db_utilsZget_connection_attributeskindr ZInvalidDatabaseTyperconnectr=r>r?r@rA) r5r6r7r8r9r:r;r<attrsrrrrDKs*9 rD.)r r index_colparams chunksizer'saferc Cs"t|dtj|||||||dS)u Return a DataFrame corresponding to the result set of the query string. Parameters ---------- sql : str SQL query. con : pg8000.Connection Use pg8000.connect() to use credentials directly or wr.postgresql.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 PostgreSQL using a Glue Catalog Connections >>> import awswrangler as wr >>> con = wr.postgresql.connect("MY_GLUE_CONNECTION") >>> df = wr.postgresql.read_sql_query( ... sql="SELECT * FROM public.my_table", ... con=con ... ) >>> con.close() rr rrFrGrHr'rI)rrBread_sql_queryrJrrrrKs2 rK) rrrrFrGrHr'rIrc Cs<|dkrd|dnd|d|d}t|||||||dS)uReturn a DataFrame corresponding the table. Parameters ---------- table : str Table name. con : pg8000.Connection Use pg8000.connect() to use credentials directly or wr.postgresql.connect() to fetch it from the Glue Catalog. schema : str, optional Name of SQL schema in database to query (if database flavor supports this). Uses default schema if None (default). 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 PostgreSQL using a Glue Catalog Connections >>> import awswrangler as wr >>> con = wr.postgresql.connect("MY_GLUE_CONNECTION") >>> df = wr.postgresql.read_sql_table( ... table="my_table", ... schema="public", ... con=con ... ) >>> con.close() NzSELECT * FROM "rr0rJ)rK) rrrrFrGrHr'rIr rrrread_sql_tables7&rLappendF) r$rrrr%r&r'r(use_column_namesrHrc  Cs.|jdkrtt|dz|} t|| ||||||d|rV|j|jjddd dgt |j } d} |rdd |j d } t j || | d } | D]>\}}d |d |d | d|}td|| ||fq|W5QRXWn8tk r(}z|t|W5d}~XYnXdS)aWrite records stored in a DataFrame into PostgreSQL. Parameters ---------- df : pandas.DataFrame Pandas DataFrame https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html con : pg8000.Connection Use pg8000.connect() to use credentials directly or wr.postgresql.connect() to fetch it from the Glue Catalog. table : str Table name schema : str Schema name mode : str Append or overwrite. 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 PostgreSQL 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 PostgreSQL using a Glue Catalog Connections >>> import awswrangler as wr >>> con = wr.postgresql.connect("MY_GLUE_CONNECTION") >>> wr.postgresql.to_sql( ... df=df, ... table="my_table", ... schema="public", ... con=con ... ) >>> con.close() Tr)r$rrrr%r&r'r()levelZinplacez, z%sr(r1)r$column_placeholdersrHz INSERT INTO "r0z" z VALUES zsql: %sN)emptyr ZEmptyDataFramerrr4Z reset_indexr&namesr2r"columnsrBZ$generate_placeholder_parameter_pairsrrZ executemanycommit Exceptionrollbackerror)r$rrrr%r&r'r(rOrHrrRZinsertion_columnsZ$placeholder_parameter_pair_generatorZ placeholders parametersr exrrrto_sql sF?       r\)NNNNNNNT)NNNNT)NNNNNT)rMFNNFrN).__doc__loggingsslrtypingrrrrrrr Zboto3ZpandaspdrZpyarrowpaZ awswranglerr r rBr Zawswrangler._configr getLogger__name__rLogger__annotations__rrCursorstrr!boolr#Z DataFrameintr4SessionrDZDataTyperKrLr\rrrrs $        O$ ;$ =