ó Ø@Qc@€sûddlmZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZmZmZdefd„ƒYZd„Zed „Zd e fd „ƒYZd efd „ƒYZdS(iÿÿÿÿ(twith_statementN(tThread(tget_connection_parameters(treplace_string_parameterstparameters_from_dsnt'dsn_parameters_to_connection_parameterst HelperExitedcB€seZd„ZRS(cC€stj||ƒ||_dS(N(t Exceptiont__init__t returncode(tselftwhatR ((s..\modules\DataMigrator.pyRs(t__name__t __module__R(((s..\modules\DataMigrator.pyRscC€sW|j}|jjdkr#d|S|jjdkr=d|Std|jjƒ‚dS(Nt MysqlNatives"%(userName)s@%(hostName)s:%(port)stMysqlNativeSockets%(userName)s@::%(socket)ss8Connection method type %s is not supported for migration(tparameterValuestdrivertnameR(tconntparam((s..\modules\DataMigrator.pytmysql_conn_string$s  cC€snt|jƒ}|jt|ƒƒ|jjp1d}t||ƒ}|rjtjdd|ƒj dƒ}n|S(NsaDRIVER={%driver%};SERVER=%host%;PORT=%port%;DATABASE={%database%};UID=%username%;PWD={%password%}s(PWD={[^;]}*|PWD=[^;]*)tt;( tdictRtupdateRRtconnectionStringTemplateRtretsubtrstrip(Rtstrip_passwordt conn_paramstconnection_string_templatet connstring((s..\modules\DataMigrator.pytodbc_conn_string-s!tTableCopyWorkercB€s,eZd„Zd„Zd„Zd„ZRS(c C€sÝtj|ƒ||_||_||_tjdd|ƒ|jjdj|ƒƒt j dkr t j |dt j dt j dt jdtd tƒ|_n9t j |dt j dt j dt jdtd tƒ|_dS( Nt MigrationsSpawning copy worker task: %st twin32tstdintstdouttstderrtuniversal_newlinestshellt close_fds(RRt_ownert result_queuet _process_argstgrtt log_debug3t send_infotjointsystplatformt subprocesstPopentPIPEtSTDOUTtTruetprocess(R townertargsR.((s..\modules\DataMigrator.pyR9s    ''cC€sx|jjƒdkr_t|ƒtkr9|jdƒ}n|jjj|ƒ|jjjƒnt d|jj ƒ‚dS(Nsutf-8s%Table copy task finished unexpectedly( R;tpolltNonettypetunicodetencodeR'twritetflushRR (R ttext((s..\modules\DataMigrator.pyt feed_inputIs c C€s,yšx´|jjƒdkr¹|jjƒr<tjdƒ‚n|jjjƒ}|dk r|j ƒj dƒ\}}}|dkr|j j ||fƒq¶|j j d|fƒqqW|jjj ƒ}|jdƒ}xv|D]n}|dk râ|j ƒj dƒ\}}}|dkr7|j j ||fƒqP|j j d|fƒqâqâW|jjd kr‰|j j d d |jjfƒn|j j dƒWn‹tjk rã}|jjd |jƒ|jƒ|j j dƒnEtk r'}ddl}|jƒ|j j d t|ƒfƒnXdS(NsCanceled by usert:tPROGRESStERRORtBEGINtENDtLOGs itDONEsWorker exited with status %ss0Copy task interrupted by user, terminating %s...t INTERRUPTEDiÿÿÿÿ(RHRIRJRK(RHRIRJRK(RMN(RNN(R;R>R?R-tquery_cancel_statusR0t UserInterruptR(treadlinetstript partitionR.tputtreadtsplitR R2Rt terminateRt tracebackt print_exctstr( R tlineR@t_tmsgtoutputtlinesteRX((s..\modules\DataMigrator.pytrunSs:     #   cC€sT|jjƒdkrC|jjƒ\}}|r”d}x?|jdƒD].}|jdƒrd|}qF|jj|ƒqFW|r”|jj |ƒq”n|r­|jj|ƒnyKt |jdƒrÒ|jj ƒn%ddl }t j|jj|jƒWqCtk r?}|jdkrq@|jjd|jj|fƒqCXn|jjƒdS(Ns s PROGRESS:RWiÿÿÿÿis!Unable to kill worker task %s: %s(R;R>R?t communicateRVt startswithR-R2R.RTthasattrRWtsignaltostkilltpidtSIGTERMtOSErrorterrnot send_errortidtwait(R toR`t last_progresstlRe((s..\modules\DataMigrator.pyRWxs,  '(R R RRFRaRW(((s..\modules\DataMigrator.pyR#8s  %t DataMigratorcB€s;eZdZd„Zd„Zd„Zd„Zd„ZRS(t wbcopytablescC€sˆt|dƒr-t|dƒr-t|dƒs3t‚||_||_||_||_||_||_g|_g|_ d|_ dS(NR2Rlt send_progress( RdtAssertionErrorR-t_optionst_src_conn_objectt _src_passwordt_tgt_conn_objectt _tgt_passwordt_taskst _processesR?t_error(R tmessage_targettoptionst srcconnobjt srcpasswordt tgtconnobjt tgtpassword((s..\modules\DataMigrator.pyR›s3        c C€sLg}tjdkrÖy•tjddtƒz}x\|jƒD]N}g}|j|dƒ|j|dƒdj|ƒ}|j|dƒq=W|jd|j ƒWdQXWqt k rÒ}t d |j ƒ‚qXn2x/|jƒD]!}|d |d|dg7}qãWd }|j s&td ƒ‚n|jjjj d krWdt|jƒg} ndt|jtƒg} |j ddg| |} |jjdj| ƒƒtjdkråtj| dtjdtjdtjdtƒ} n*tj| dtjdtjdtjƒ} |jdjdƒ} x}| jƒdkr¤| j| ƒ\} }d} | rh|| 7}n|r›x*|jdƒD]}|jj|ƒq~Wnd}q(W| j dkr<xg|jdƒD](}|j!dƒrÇ|jdƒd^qÇD]A\}}}t"|j#ƒƒ}||7}|||d|dR?RbRVR RctintRR(R t working_sett table_paramt table_filettasktfieldsR[R`R(R=targvtoutt passwordsRoRqttotaltschemattabletcount((s..\modules\DataMigrator.pytcount_table_rows©sV 3*  N cC€sÒg}tjdkr+yêtjddtƒÏ}x±|jƒD]£}g}|j|dƒ|j|dƒ|j|dƒ|j|dƒ|jddƒr³|j|dƒn |jd ƒd j |ƒ}|j |d ƒq=W|jd |j ƒWdQXWqžt k r'}t d |jƒ‚qžXnsxp|jƒD]b}|d|d|d|d|dg7}|jddƒr|j|dƒq8|jd ƒq8Wd} t|ƒ|krÅt|ƒ}n|jƒ} | ddg7} |jjdtƒr| jdƒn|jjdtƒr+| jdƒn| jdt|ƒƒ|jg| |} ||_tjt|ƒƒ|_t|j| |jƒ} | j|jd |jd ƒ| jƒ|jƒ} | jƒ| S(NR&R„R…R†R‡t target_schemat target_tabletselect_expressiont*s s s--table-file=%ssError creating table file: %ss--tableRs --progresss--passwords-from-stdintTruncateTargetTabless--truncate-targettDebugTableCopys--log-level=debug3s--thread-count=(R4R5RRŽRRR‘tgetR?R3RCRR’RR“tlenthelper_basic_arglistRvRZR”t _working_settQueuet _result_queueR#R-RFRxRztstarttprocess_until_doneRW(R t num_processesR—R˜R™RšR›R[R`R(R=Rœtworkertresults((s..\modules\DataMigrator.pyt migrate_dataãsT )     cC€s|jjjjdkr1dt|jƒg}ndt|jtƒg}|dt|jƒg7}|jjj dt ƒrŒ|j dƒn|S(NRˆs--mysql-source=%ss--odbc-source=%ss --target=%stODBCDriverUsesUTF8s--force-utf8-for-source( RwRR<RRR"R:RyRRªRR‘(R R=((s..\modules\DataMigrator.pyR¬sc C€sôd}x%|jjƒD]}||d7}qWi}t|_tƒ}t}x¡trï|ry|jjƒ\}}Wq¤tj k r‹Pq¤Xn|jj ƒ\}}|dkrã|j dƒd}|j |ƒ|j j|ƒqO|dkrW|j dƒd}||kr|j|ƒn|j j|ƒt|j |tdfƒdf||s        `