o ;Dia`@s|ddlmZmZmZddlZddlmZmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZddlmZmZeeZejZejZdd ZGd d d e Zd d Z ddZ!ddZ"ddZ#ddZ$ddZ%   d1ddZ&e&e _&d2ddZ'dZ(d Z)  d2d!d"Z*  d2d#d$Z+  d2d%d&Z,  d2d'd(Z-  d2d)d*Z.  d2d+d,Z/  d2d-d.Z0d3d/d0Z1e1e _1dS)4)absolute_importprint_functiondivisionN)next text_type string_types) ArgumentError)Table)_is_dbapi_connection_is_dbapi_cursor_is_sqlalchemy_connection_is_sqlalchemy_engine_is_sqlalchemy_session_is_clikchouse_dbapi_connection_quote _placeholders) drop_table create_tablecOs4t|trddl}||}t||g|Ri|S)a Provides access to data from any DB-API 2.0 connection via a given query. E.g., using :mod:`sqlite3`:: >>> import petl as etl >>> import sqlite3 >>> connection = sqlite3.connect('example.db') >>> table = etl.fromdb(connection, 'SELECT * FROM example') E.g., using :mod:`psycopg2` (assuming you've installed it first):: >>> import petl as etl >>> import psycopg2 >>> connection = psycopg2.connect('dbname=example user=postgres') >>> table = etl.fromdb(connection, 'SELECT * FROM example') E.g., using :mod:`pymysql` (assuming you've installed it first):: >>> import petl as etl >>> import pymysql >>> connection = pymysql.connect(password='moonpie', database='thangs') >>> table = etl.fromdb(connection, 'SELECT * FROM example') The `dbo` argument may also be a function that creates a cursor. N.B., each call to the function should return a new cursor. E.g.:: >>> import petl as etl >>> import psycopg2 >>> connection = psycopg2.connect('dbname=example user=postgres') >>> mkcursor = lambda: connection.cursor(cursor_factory=psycopg2.extras.DictCursor) >>> table = etl.fromdb(mkcursor, 'SELECT * FROM example') The parameter `dbo` may also be an SQLAlchemy engine, session or connection object. The parameter `dbo` may also be a string, in which case it is interpreted as the name of a file containing an :mod:`sqlite3` database. Note that the default behaviour of most database servers and clients is for the entire result set for each query to be sent from the server to the client. If your query returns a large result set this can result in significant memory usage at the client. Some databases support server-side cursors which provide a means for client libraries to fetch result sets incrementally, reducing memory usage at the client. To use a server-side cursor with a PostgreSQL database, e.g.:: >>> import petl as etl >>> import psycopg2 >>> connection = psycopg2.connect('dbname=example user=postgres') >>> table = etl.fromdb(lambda: connection.cursor(name='arbitrary'), ... 'SELECT * FROM example') For more information on server-side cursors see the following links: * http://initd.org/psycopg/docs/usage.html#server-side-cursors * http://mysql-python.sourceforge.net/MySQLdb.html#using-and-extending rN) isinstancersqlite3connectDbView)dboqueryargskwargsrrC/var/www/Datamplify/venv/lib/python3.10/site-packages/petl/io/db.pyfromdbs = rc@seZdZddZddZdS)rcOs||_||_||_||_dSN)rrrr)selfrrrrrrr__init__^s zDbView.__init__cCst|jrtd|jt}nOt|jrtdtdt}n?t|jr,td|jt}n1t |jr:td|jt }n#t |jrHtd|jt }nt |jrVtd|jt}ntd|j||j|jg|jRi|jS) N-assuming %r is standard DB-API 2.0 connection)assuming %r is standard DB-API 2.0 cursorzxusing a DB-API cursor with fromdb() is not recommended and may lead to unexpected results, a DB-API connection is better5assuming %r instance of sqlalchemy.engine.base.Engine6assuming %r instance of sqlalchemy.orm.session.Session9assuming %r instance of sqlalchemy.engine.base.Connectionz,assuming %r is a function returning a cursor$unsupported database object type: %r)r rdebug_iter_dbapi_connectionr warning_iter_dbapi_cursorr _iter_sqlalchemy_enginer_iter_sqlalchemy_sessionr _iter_sqlalchemy_connectioncallable_iter_dbapi_mkcursrrrr)r _iterrrr__iter__ds6         zDbView.__iter__N)__name__ __module__ __qualname__r!r2rrrrr\s rcosF|}zt||g|Ri|D]}|VqW|dS|wr)r+close)mkcursrrrcursorrowrrrr0sr0cosH|}zt||g|Ri|D]}|VqW|dS|wr)r8r+r6) connectionrrrr8r9rrrr)sr)cos|j|g|Ri|zt|}Wnty"t|}Ynwzt|}Wn ty4d}Ynwdd|jD}t|V|durHdS|V|D]}|VqMdS)NcSsg|]}|dqS)rr).0drrr z&_iter_dbapi_cursor..)executeiter TypeErrorfetchallr StopIteration descriptiontuple)r8rrrit first_rowhdrr9rrrr+s(     r+cos:|}t||g|Ri|D]}|Vq|dSr)rr.r6)enginerrrr:r9rrrr,s  r,cosJtd||j|g|Ri|}|}t|V|D]}|VqdS)Nconnection: %r)r(r?keysrE)r:rrrresultsrHr9rrrr.s  r.cos@|j|g|Ri|}|}t|V|D]}|VqdSr)r?rKrE)sessionrrrrLrHr9rrrr-s r-TFc Csd} t|trddl} | |}d} z.|r,|rt||||dt|||||||| | d t|||||ddW| r?|dSdS| rH|ww)a Load data into an existing database table via a DB-API 2.0 connection or cursor. Note that the database table will be truncated, i.e., all existing rows will be deleted prior to inserting the new data. E.g.:: >>> import petl as etl >>> table = [['foo', 'bar'], ... ['a', 1], ... ['b', 2], ... ['c', 2]] >>> # using sqlite3 ... import sqlite3 >>> connection = sqlite3.connect('example.db') >>> # assuming table "foobar" already exists in the database ... etl.todb(table, connection, 'foobar') >>> # using psycopg2 >>> import psycopg2 >>> connection = psycopg2.connect('dbname=example user=postgres') >>> # assuming table "foobar" already exists in the database ... etl.todb(table, connection, 'foobar') >>> # using pymysql >>> import pymysql >>> connection = pymysql.connect(password='moonpie', database='thangs') >>> # tell MySQL to use standard quote character ... connection.cursor().execute('SET SQL_MODE=ANSI_QUOTES') >>> # load data, assuming table "foobar" already exists in the database ... etl.todb(table, connection, 'foobar') N.B., for MySQL the statement ``SET SQL_MODE=ANSI_QUOTES`` is required to ensure MySQL uses SQL-92 standard quote characters. A cursor can also be provided instead of a connection, e.g.:: >>> import psycopg2 >>> connection = psycopg2.connect('dbname=example user=postgres') >>> cursor = connection.cursor() >>> etl.todb(table, cursor, 'foobar') The parameter `dbo` may also be an SQLAlchemy engine, session or connection object. The parameter `dbo` may also be a string, in which case it is interpreted as the name of a file containing an :mod:`sqlite3` database. If ``create=True`` this function will attempt to automatically create a database table before loading the data. This functionality requires `SQLAlchemy `_ to be installed. **Keyword arguments:** table : table container Table data to load dbo : database object DB-API 2.0 connection, callable returning a DB-API 2.0 cursor, or SQLAlchemy connection, engine or session tablename : string Name of the table in the database schema : string Name of the database schema to find the table in commit : bool If True commit the changes create : bool If True attempt to create the table before loading, inferring types from a sample of the data (requires SQLAlchemy) drop : bool If True attempt to drop the table before recreating (only relevant if create=True) constraints : bool If True use length and nullable constraints (only relevant if create=True) metadata : sqlalchemy.MetaData Custom table metadata (only relevant if create=True) dialect : string One of {'access', 'sybase', 'sqlite', 'informix', 'firebird', 'mysql', 'oracle', 'maxdb', 'postgresql', 'mssql'} (only relevant if create=True) sample : int Number of rows to sample when inferring types etc. Set to 0 to use the whole table (only relevant if create=True) .. note:: This function is in principle compatible with any DB-API 2.0 compliant database driver. However, at the time of writing some DB-API 2.0 implementations, including cx_Oracle and MySQL's Connector/Python, are not compatible with this function, because they only accept a list argument to the cursor.executemany() function called internally by :mod:`petl`. This can be worked around by proxying the cursor objects, e.g.:: >>> import cx_Oracle >>> connection = cx_Oracle.Connection(...) >>> class CursorProxy(object): ... def __init__(self, cursor): ... self._cursor = cursor ... def executemany(self, statement, parameters, **kwargs): ... # convert parameters to a list ... parameters = list(parameters) ... # pass through to proxied cursor ... return self._cursor.executemany(statement, parameters, **kwargs) ... def __getattr__(self, item): ... return getattr(self._cursor, item) ... >>> def get_cursor(): ... return CursorProxy(connection.cursor()) ... >>> import petl as etl >>> etl.todb(tbl, get_cursor, ...) Note however that this does imply loading the entire table into memory as a list prior to inserting into the database. FrNT)schemacommit)rOrP constraintsmetadatadialectsamplerOrPtruncate)rrrrrr_todbr6) tabler tablenamerOrPcreatedroprQrRrSrT needs_closingrrrrtodbs*v      r]cCs0t|rtd|t||||||ddSt|r*td|t||||||ddSt|r>tdt||||||ddSt|rStd|t||||||ddSt |rhtd|t ||||||ddSt |r}td|t ||||||ddSt |rtd|t||||||ddStd |) Nz/assuming %r is clickhosue DB-API 2.0 connectionrUr"r#r$r%r&zFassuming %r is a function returning standard DB-API 2.0 cursor objectsr')rr(!_todb_clikchouse_dbapi_connectionr _todb_dbapi_connectionr _todb_dbapi_cursorr _todb_sqlalchemy_enginer_todb_sqlalchemy_sessionr _todb_sqlalchemy_connectionr/_todb_dbapi_mkcursr)rXrrYrOrPrVrrrrWasL                   rWzDELETE FROM %szINSERT INTO %s (%s) VALUES (%s)cCst|}|durt|d|}td|t|}t|}ttt|}dd|D} td| t|| } td| |} |rVt |} td| | | | |} d | } t || | f}td || ||td | |rtd |dSdS) N. tablename: %rcSg|]}t|qSrrr;nrrrr=r>z*_todb_dbapi_connection..column names: %rplaceholders: %rtruncate the table via query %r, insert data via query %rclose the cursorcommit transaction)rr(r@rlistmaprrr8SQL_TRUNCATE_QUERYr?r6joinSQL_INSERT_QUERY executemanyrPrXr:rYrOrPrVrFrHfldscolnames placeholdersr8 truncatequeryinsertcolnames insertqueryrrrr_s8          r_cCst|}|durt|d|}td|t|}t|}ttt|}dd|D} td| t|| } td| |} |rVd|} td| | | | |} d | } d || f}td || ||td | |rtd | dSdS)NrerfcSrgrrhrirrrr=r>z5_todb_clikchouse_dbapi_connection..rkrlzTRUNCATE TABLE IF EXISTS %srmrnzINSERT INTO %s (%s) VALUESrorprq)rr(r@rrrrsrrr8r?r6rurwrPrxrrrr^s8           r^cCst|}|durt|d|}td|t|}t|}ttt|}dd|D} td| td|} t| ds@Jd| j} t | | } td | |rdt |} td | | | | |} d | }t||| f}td || ||| |rtd | dSdS)NrerfcSrgrrhrirrrr=r>z&_todb_dbapi_mkcurs..rkzobtain cursor and connectionr:&could not obtain connection via cursorrlrmrnrorq)rr(r@rrrrsrhasattrr:rrtr?r6rurvrwrP)rXr7rYrOrPrVrFrHryrzr8r:r{r|r}r~rrrrds@           rdcCst|}|durt|d|}td|t|}t|}ttt|}dd|D} td| tdt|ds=Jd|j} t | | } td | |rZt |} td | | | d | } t || | f}td |||||r~td | dSdS)NrerfcSrgrrhrirrrr=3r>z&_todb_dbapi_cursor..rkzobtain connection via cursorr:rrlrmrnrorq)rr(r@rrrrsrrr:rrtr?rurvrwrP)rXr8rYrOrPrVrFrHryrzr:r{r|r}r~rrrr`&s8           r`cCt||||||ddSNrU)rcr)rXrIrYrOrPrVrrrraSs  racCstd|t|}|durt|d|}td|t|}t|}ttt|}dd|D} td| |j} | j} t| | } td| |rOtd| } |r_t |}td || |d | }t ||| f}td ||D]}| ||qs|rtd | dSdS) NrJrerfcSrgrrhrirrrr=ir>z/_todb_sqlalchemy_connection..rkrlzbegin transactionrmrnrorq)r(rr@rrrrsrr:rbeginrtr?rurvrP)rXr:rYrOrPrVrFrHryrzproxied_raw_connectionactual_raw_connectionr{transr|r}r~r9rrrrcZs<          rccCrr)rcr:)rXrMrYrOrPrVrrrrbs  rbcCs`d}t|trddl}||}d}zt|||||ddW|r&|dSdS|r/|ww)a  Load data into an existing database table via a DB-API 2.0 connection or cursor. As :func:`petl.io.db.todb` except that the database table will be appended, i.e., the new data will be inserted into the table, and any existing rows will remain. FrNTrU)rrrrrWr6)rXrrYrOrPr\rrrrappenddbs     r)NTFFTNNrN)NTF)NT)2 __future__rrrlogging petl.compatrrr petl.errorsrpetl.util.baser petl.io.db_utilsr r r r rrrrpetl.io.db_createrr getLoggerr3loggerr(r*rrr0r)r+r,r.r-r]rWrtrvr_r^rdr`rarcrbrrrrrsd  ( D4     3 / / 0 -  5