a g g*@sddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddlm Z ddlm Z dd l m Z dd l mZd d Zd2ddZddZddZddZd3ddZd4ddZd5ddZd6d!d"Zd#d$Zd%d&Zd7d(d)Zd*d+Zd,d-Zd.d/Zd0d1ZdS)8N)config) provision)util)command)script)Script)ScriptDirectorycCstjrdtjSdSdS)Nz scratch_%sZscratch)rZFOLLOWER_IDENTr r V/var/www/html/llm_bihealth/app/venv/lib/python3.9/site-packages/alembic/testing/env.py_get_staging_directorys r TgenericFcCst}|rtjtd}tj|r2Jd|tj|||d|rzt |dWnt yhYn0|dvszJ|t tj|dd|vrdndt j |}|S)Nscriptsz2staging directory %s already exists; poor cleanup?)templateenv.py)Zpep3147_envonlysimplepep3147_everythingpep3147r)_testing_configospathjoinr existsrinitrZload_python_fileAttributeErrormake_sourcelessrr from_config)creater sourcelesscfgrZscr r r staging_envs*    r!cCs(ddlm}|jttddS)NrenginesT)sqlalchemy.testingr#Ztesting_reaperZ close_allshutilrmtreer r"r r r clear_staging_env<s  r'cCsVtjtd}tj|d}t|d}||Wdn1sH0YdS)Nrzscript.py.makow)rrrr openwrite)txtdir_rfr r r script_file_fixtureCs r.cCsvtjtd}d|}tj|d}t|}|r>t|t|d}||Wdn1sh0YdS)Nrz6 from alembic import context config = context.config rr() rrrr rpyc_file_from_pathunlinkr)r*)r+r,rpyc_pathr-r r r env_file_fixtureJs    r2foo.dbcKs>tjtd}d||f}|r.tjr.||d<tj|||dS)Nrzsqlite:///%s/%sscope)urlfutureoptions)rrrr rZsqla_14 testing_utilZtesting_engine)tempnamer6r4r7r,r5r r r _sqlite_file_db^s   r:cCsNtjtd}d|}|p(dtjjjv}td|||r:dnd|rDdndfS) Nrsqlite:///%s/foo.dbr6a [alembic] script_location = %s sqlalchemy.url = %s sourceless = %s %s [loggers] keys = root,sqlalchemy [handlers] keys = console [logger_root] level = WARN handlers = console qualname = [logger_sqlalchemy] level = DEBUG handlers = qualname = sqlalchemy.engine [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S truefalsezsqlalchemy.future = true rrrr rdb __class__ __module___write_config_file)rr6r,r5sqlalchemy_futurer r r _sqlite_testing_configfs%  rEr>cCsLtjtd}dtjjjv}d|}td|||r6dnd|r@dnd|fS)Nrr6r;a [alembic] script_location = %s sqlalchemy.url = %s sqlalchemy.future = %s sourceless = %s version_locations = %%(here)s/model1/ %%(here)s/model2/ %%(here)s/model3/ %s [loggers] keys = root [handlers] keys = console [logger_root] level = WARN handlers = console qualname = [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S r<r=r?)rZextra_version_locationr,rDr5r r r _multi_dir_testing_configs!  rF postgresqlcCs"tjtd}td|||fS)zLuse a postgresql url with no host so that connections guaranteed to failra [alembic] script_location = %s sqlalchemy.url = %s:// %s [loggers] keys = root [handlers] keys = console [logger_root] level = WARN handlers = console qualname = [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S )rrrr rC)dialectZ directivesr,r r r _no_sql_testing_configs rIcCs@t}t|jd}||Wdn1s20Y|S)Nr()rr)Zconfig_file_namer*)textr r-r r r rCs(rCcCs<ddlm}tttjs(tt|tjtdS)NrConfigztest_alembic.ini) Zalembic.configrLraccessr F_OKmkdirrrrKr r r rs  rasciic Cs|j|}|j}t|}|r*||}t|d}||Wdn1sT0Yt |}|rvt |t ||} |j| j}|j| jkrtd|jj| dd|rt||dkrdnddS)Nwbz2Can't change down_revision on a refresh operation.T)_replacerrr)Z revision_map get_revisionrtextwrapdedentencoder)r*rr/rr0r Z _from_pathrevisionZ down_revision ExceptionZ add_revisionr) Z scriptdirrev_idcontentencodingroldrfpr1rr r r write_scripts*    (    r^cCsddl}|||dkrZt|}tjjd}tj |\}}||}t |||}n|dvsfJt|}t |tj sJt|dS)Nrr)rr) py_compilecompilerr/ importlib machineryBYTECODE_SUFFIXESrrsplitextr%moverMrNr0)rstyler_r1suffixfilepathextZsimple_pyc_pathr r r rs      rcCst}t}t}t|}|j|ddddt||d||j|dd|dt||d|d|d d d |j|d d|dt||d ||f|||fS)Nz revision aTbase)refreshheadz"Rev A" revision = '%s' down_revision = None from alembic import op def upgrade(): op.execute("CREATE STEP 1") def downgrade(): op.execute("DROP STEP 1") z revision bu/# coding: utf-8 "Rev B, méil, %3" revision = 'z' down_revision = 'z}' from alembic import op def upgrade(): op.execute("CREATE STEP 2") def downgrade(): op.execute("DROP STEP 2") zutf-8)r[z revision cz"Rev C" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 3") def downgrade(): op.execute("DROP STEP 3") rrYr rZgenerate_revisionr^)r abcrr r r three_rev_fixture4s@ rqcCst}t}t}t|}|j|d|dddt||d||f|j|d|dddt||d||f|j|d|dddt||d||f|||fS) z:Create a multiple head fixture from the three-revs fixturezrevision d from bT)rlZsplicerkz"Rev D" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 4") def downgrade(): op.execute("DROP STEP 4") zrevision e from dz"Rev E" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 5") def downgrade(): op.execute("DROP STEP 5") zrevision f from bz"Rev F" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 6") def downgrade(): op.execute("DROP STEP 6") rm)r rnrorpder-rr r r multi_heads_fixturesF    rtcCs`tjtd}dtjjjv}d|}ddd| D}t d||rRdnd ||fS) z?alembic.ini fixture to work exactly with the 'multidb' templaterr6z, z css |]\}}d||jfVqdS)z[%s] sqlalchemy.url = %sN)r5).0keyvaluer r r sz*_multidb_testing_config..a [alembic] script_location = %s sourceless = false sqlalchemy.future = %s databases = %s %s [loggers] keys = root [handlers] keys = console [logger_root] level = WARN handlers = console qualname = [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S r<r=) rrrr rr@rArBkeysitemsrC)r#r,rDZ databasesr r r _multidb_testing_configs  r{)TrF)r3FN)FF)Fr>)rGr>)rPF)Zimportlib.machineryrarr%rTr$rrr>rr8rrr r r r!r'r.r2r:rErFrIrCrr^rrqrtr{r r r r s8         $  4 1 ( MY