Edit File: env.cpython-37.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)�alembicr r �_testing_config�os�path�joinr �exists�shutil�rmtree�initr Zload_python_file�AttributeError�make_sourcelessr � from_config)�creater � 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)�tempnamer' r1 r r r �_sqlite_file_dbJ 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 [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' r1 r r r �_sqlite_testing_configP s r7 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 r4 r5 )r r r r r6 )r r' r1 r r r �_multi_dir_testing_configv s r8 � 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 r6 )ZdialectZ directivesr' r r r �_no_sql_testing_config� s r; c C s, t � }t|jd��}|�| � W d Q R X |S )Nr# )r r$ Zconfig_file_namer% )�textr r( r r r r6 � s r6 c C s< ddl m} t�t� tj�s(t�t� � | tj�t� d��S )Nr )�Configztest_alembic.ini) Zalembic.configr= r r+ r r, �mkdirr r )r= 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_path�revisionZ down_revision� ExceptionZ add_revisionr ) Z scriptdir�rev_id�content�encodingr �oldr �fpr. r r r r � write_script� s" rL 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 �copyfiler- )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)rI 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 rG r r Zgenerate_revisionrL r )r �a�b�cr r r r �three_rev_fixture� s rR 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)r1 )�.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�itemsr6 )r r' Z databasesr r r �_multidb_testing_config- s rY )Tr F)r0 )F)F)r9 r: )r? F)r r rB Z util.compatr r r r r: r r r r r! r" r) r/ r3 r7 r8 r; r6 r rL r rR rY r r r r �<module> s, & % % 8