관리-도구
편집 파일: env.cpython-38.pyc
U 0?�f* � @ s d dl Zd dlZd dlZd dlZd dlmZ d dlmZ ddlm Z ddlm Z ddlm Z ddlm Z dd l m Z dd l mZ d d � Zd2dd�Zdd� Zdd� Zdd� Zd3dd�Zd4dd �Zd5d d �Zd6d!d"�Zd#d$� Zd%d&� Zd7d(d)�Zd*d+� Zd,d-� Z d.d/� Z d0d1� Z dS )8� N)�config)� provision� )�util� )�command)�script)�Script)�ScriptDirectoryc C s t jrdt j S dS d S )Nz scratch_%sZscratch)r ZFOLLOWER_IDENT� r r �A/opt/hc_python/lib/python3.8/site-packages/alembic/testing/env.py�_get_staging_directory s r T�genericFc C s� t � }| r�tj�t� d�}tj�|�r2td| ��tj|||d� |r�zt � |d� W n t k rj Y nX |dks|t|��t tj�|d�d|kr�dnd� t j�|�}|S )N�scriptsz2staging directory %s already exists; poor cleanup?)�template�env.py)Zpep3147_envonly�simple�pep3147_everything�pep3147r )�_testing_config�os�path�joinr �exists�AssertionErrorr �initr Zload_python_file�AttributeError�make_sourcelessr r � from_config)�creater � sourceless�cfgr �scr r r � staging_env s* � � � r# c C s( ddl m} | j�� t�t� d� d S )Nr ��enginesT)�sqlalchemy.testingr% Ztesting_reaperZ close_all�shutil�rmtreer r$ r r r �clear_staging_env: s r) c C sB t j�t� d�}t j�|d�}t|d��}|�| � W 5 Q R X d S )Nr zscript.py.mako�w)r r r r �open�write)�txt�dir_r �fr r r �script_file_fixtureA s r0 c C sb t j�t� d�}d| } t j�|d�}t�|�}|r>t �|� t|d��}|�| � W 5 Q R X d S )Nr z6 from alembic import context config = context.config r r* ) r r r r r �pyc_file_from_path�unlinkr+ r, )r- r. r �pyc_pathr/ r r r �env_file_fixtureH s �� r4 �foo.dbc K s> t j�t� d�}d|| f }|r.tjr.||d< tj|||d�S )Nr zsqlite:///%s/%s�scope)�url�future�options)r r r r r Zsqla_14� testing_utilZtesting_engine)�tempnamer8 r6 r9 r. r7 r r r �_sqlite_file_db\ s r< c C sN t j�t� d�}d| }|p(dtjjjk}td||| r:dnd|rDdndf �S ) Nr �sqlite:///%s/foo.dbr8 a� [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 �true�falsezsqlalchemy.future = true� � r r r r r �db� __class__� __module__�_write_config_file)r r8 r. r7 �sqlalchemy_futurer r r �_sqlite_testing_configd s % ���rG r@ c C sL t j�t� d�}dtjjjk}d| }td|||r6dnd| r@dnd|f �S )Nr r8 r= 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? rA )r Zextra_version_locationr. rF r7 r r r �_multi_dir_testing_config� s ! ���rH � postgresqlc C s" t j�t� d�}td|| |f �S )zLuse a postgresql url with no host so that connections guaranteed to failr a� [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 )r r r r rE )�dialectZ directivesr. r r r �_no_sql_testing_config� s ��rK c C s, t � }t|jd��}|�| � W 5 Q R X |S )Nr* )r r+ Zconfig_file_namer, )�textr! r/ r r r rE � s rE c C s< ddl m} t�t� tj�s(t�t� � | tj�t� d��S )Nr ��Configztest_alembic.ini) Zalembic.configrN r �accessr �F_OK�mkdirr r rM r r r r � s r �asciic C s� | j �|�}|j}t�|�}|r*|�|�}t|d��}|�|� W 5 Q R X t� |�}|rbt � |� t � | |�} | j �| j�}|j| jkr�td��| j j| dd� |r�t||dkr�dnd� d S )N�wbz2Can't change down_revision on a refresh operation.T)�_replacer r r )Z revision_map� get_revisionr �textwrap�dedent�encoder+ r, r r1 r r2 r Z _from_path�revisionZ down_revision� ExceptionZ add_revisionr ) Z scriptdir�rev_id�content�encodingr �oldr �fpr3 r r r r � write_script s, � �r` c C s� dd l }|�| � |dkrZt�| �}tjjd }tj� | �\}}|| }t � ||� |}n|dksft �t�| �}t� |tj�s�t �t�| � d S )Nr r )r r )� py_compile�compiler r1 � importlib� machinery�BYTECODE_SUFFIXESr r �splitextr' �mover rO rP r2 )r �stylera r3 �suffix�filepath�extZsimple_pyc_pathr r r r s r c C s� t �� }t �� }t �� }t�| �}|j|dddd� t||d| � |j|dd|d� t||d|� d|� d �d d � |j|d d|d� t||d ||f � |||fS )Nz revision aT�base)�refresh�headz�"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") �r r[ r r Zgenerate_revisionr` )r! �a�b�cr r r r �three_rev_fixture1 s@ �������rs c C s� t �� }t �� }t �� }t�| �}|j|d|ddd� t||d||f � |j|d|ddd� t||d||f � |j|d|ddd� t||d||f � |||fS ) z:Create a multiple head fixture from the three-revs fixturezrevision d from bT)rn Zsplicerm z�"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") ro )r! rp rq rr �d�er/ r r r r �multi_heads_fixture~ s^ ��� ��� ���rv c C s` t j�t� d�}dtjjjk}d�| �� �}d�dd� | � � D ��} t d||rRdnd || f �S ) z?alembic.ini fixture to work exactly with the 'multidb' templater r8 z, z c s s | ]\}}d ||j f V qdS )z[%s] sqlalchemy.url = %sN)r7 )�.0�key�valuer r r � <genexpr>� s �z*_multidb_testing_config.<locals>.<genexpr>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? ) r r r r r rB rC rD �keys�itemsrE )r% r. rF Z databasesr r r �_multidb_testing_config� s � ��r} )Tr F)r5 FN)FF)Fr@ )rI r@ )rR F) Zimportlib.machineryrc r r' rV r& r r r@ r r: r r r r r r# r) r0 r4 r< rG rH rK rE r r` r rs rv r} r r r r �<module> s: " 4 1 ( � MY