Edit File: env.cpython-37.opt-1.pyc
B �M!VU � @ s� d dl Z d dlZd dlZddlmZ ddlmZmZ ddlm Z ddlm Z ddlm Z d d � Z d.dd�Z dd� Zdd� Zdd� Zd/dd�Zd0dd�Zd1dd �Zd2d d �Zd!d"� Zd#d$� Zd3d&d'�Zd(d)� Zd*d+� Zd,d-� ZdS )4� N� )�u)�Script�ScriptDirectory)�util� )�engines)� provisionc C s t jrdt j S dS d S )Nz scratch_%sZscratch)r ZFOLLOWER_IDENT� r r �D/opt/alt/python37/lib/python3.7/site-packages/alembic/testing/env.py�_get_staging_directory s r T�genericFc C s� ddl m}m} t� }| r�tj�t� d�}tj�|�r@t � |� |j |||d� |r�yt � |d� W n tk rx Y nX ttj�|d�� |j�|�}|S )Nr )�command�script�scripts)�templatezenv.py)Zalembicr r �_testing_config�os�path�joinr �exists�shutil�rmtreeZinitr Zload_python_file�AttributeError�make_sourcelessr � from_config)Zcreater � sourcelessr r �cfgr Zscr r r � staging_env s r c C s t �t� d� d S )NT)r r r r r r r �clear_staging_env. s r c C sB t j�t� d�}t j�|d�}t|d��}|�| � W d Q R X d S )Nr zscript.py.mako�w)r r r r �open�write)�txt�dir_r �fr r r �script_file_fixture2 s r& c C sl t j�t� d�}d| } t j�|d�}t�|�}t �|t j�rHt �|� t |d��}|� | � W d Q R X d S )Nr z6 from alembic import context config = context.config zenv.pyr ) r r r r r �pyc_file_from_path�access�F_OK�unlinkr! r" )r# r$ r �pyc_pathr% r r r �env_file_fixture9 s r, �foo.dbc C s( t j�t� d�}d|| f }tj|d�S )Nr zsqlite:///%s/%s)�url)r r r r r Ztesting_engine)Ztempnamer$ r. r r r �_sqlite_file_dbJ s r/ c C s2 t j�t� d�}d| }td||| r(dndf �S )Nr zsqlite:///%s/foo.dba� [alembic] script_location = %s sqlalchemy.url = %s sourceless = %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 �true�false)r r r r �_write_config_file)r r$ r. r r r �_sqlite_testing_configP s r3 c C s2 t j�t� d�}d| }td||| r(dndf �S )Nr zsqlite:///%s/foo.dba� [alembic] script_location = %s sqlalchemy.url = %s sourceless = %s version_locations = %%(here)s/model1/ %%(here)s/model2/ %%(here)s/model3/ [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 r0 r1 )r r r r r2 )r r$ r. r r r �_multi_dir_testing_configv s r4 � postgresql� c 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 r2 )ZdialectZ directivesr$ r r r �_no_sql_testing_config� s r7 c C s, t � }t|jd��}|�| � W d Q R X |S )Nr )r r! Zconfig_file_namer" )�textr r% r r r r2 � s r2 c C s< ddl m} t�t� tj�s(t�t� � | tj�t� d��S )Nr )�Configztest_alembic.ini) Zalembic.configr9 r r( r r) �mkdirr r )r9 r r r r � s r �asciic C s� | j �|�}|j}t�|�}|r*|�|�}t|d��}|�|� W d Q R X t� |�}t � |t j �rlt � |� t�| |�} | j �| j�}|j| jkr�td��| j j| dd� |r�t|� d S )N�wbz2Can't change down_revision on a refresh operation.T)�_replace)Z revision_mapZ get_revisionr �textwrap�dedent�encoder! r" r r' r r( r) r* r Z _from_pathZrevisionZ down_revision� ExceptionZ add_revisionr ) Z scriptdir�rev_idZcontent�encodingr �oldr �fpr+ r r r r � write_script� s" rF c C sN t �| �}t�|tj�s t�t �| �}t�|tj�s@t�||� t� | � d S )N) r r' r r( r) �AssertionErrorZsimple_pyc_file_from_pathr Zcopyfiler* )r r+ Zsimple_pyc_pathr r r r � s r c C s� t �� }t �� }t �� }t�| �}|j|ddd� t||d| � |j|ddd� t||td�||f dd� |j|d dd� t||d ||f � |||fS ) Nz revision aT)Zrefreshz�"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" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 2") def downgrade(): op.execute("DROP STEP 2") zutf-8)rC 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 rB r r Zgenerate_revisionrF r )r �a�b�cr r r r �three_rev_fixture� s rK c C sH t j�t� d�}d�| �� �}d�dd� | �� D ��} td||| f �S )z?alembic.ini fixture to work exactly with the 'multidb' templater z, z c s s | ]\}}d ||j f V qdS )z[%s] sqlalchemy.url = %sN)r. )�.0�key�valuer r r � <genexpr>6 s z*_multidb_testing_config.<locals>.<genexpr>a� [alembic] script_location = %s sourceless = false 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 �keys�itemsr2 )r r$ Z databasesr r r �_multidb_testing_config- s rR )Tr F)r- )F)F)r5 r6 )r; F)r r r> Z util.compatr r r r r6 r r r r r r r& r, r/ r3 r4 r7 r2 r rF r rK rR r r r r �<module> s, & % % 8