o
    8c*                     @   s  d dl Z d dlZd dlZd dlmZ d dlmZ ddlmZ ddlmZ ddl	m
Z
 ddl	mZ dd	lmZ dd
lmZ ddlmZ dd Zd3ddZdd Zdd Zdd Zd4ddZd5ddZd6dd Zd7d"d#Zd$d% Zd&d' Z	d8d)d*Zd+d, Zd-d. Zd/d0 Zd1d2 ZdS )9    N)config)	provision   )util   )Script)ScriptDirectory)get_current_bytecode_suffixes)has_pep3147)uc                   C   s   t jrdt j S dS )Nz
scratch_%sscratch)r   FOLLOWER_IDENT r   r   I/var/www/html/gps/gps/lib/python3.10/site-packages/alembic/testing/env.py_get_staging_directory   s   
r   TgenericFc                 C   s   ddl m}m} t }| rUtjt d}tj|r!J d| |j	|||d |rUzt
|d W n	 ty<   Y nw |dv sEJ |ttj|dd|v rRdnd	 |j|}|S )
Nr   )commandscriptscriptsz2staging directory %s already exists; poor cleanup?)templateenv.py)pep3147_envonlysimplepep3147_everythingpep3147r   )alembicr   r   _testing_configospathjoinr   existsinitr   load_python_fileAttributeErrormake_sourcelessr   from_config)creater   
sourcelessr   r   cfgr   scr   r   r   staging_env   s.   
r*   c                  C   s(   ddl m}  | j  tt d d S )Nr   enginesT)sqlalchemy.testingr,   testing_reaper	close_allshutilrmtreer   r+   r   r   r   clear_staging_env?   s   
r2   c                 C   sX   t jt d}t j|d}t|d}||  W d    d S 1 s%w   Y  d S )Nr   zscript.py.makow)r   r   r   r   openwrite)txtdir_r   fr   r   r   script_file_fixtureF   s
   "r9   c                 C   sx   t jt d}d|  } t j|d}t|}|rt | t|d}||  W d    d S 1 s5w   Y  d S )Nr   z6
from alembic import context

config = context.config
r   r3   )	r   r   r   r   r   pyc_file_from_pathunlinkr4   r5   )r6   r7   r   pyc_pathr8   r   r   r   env_file_fixtureM   s   	

"r=   foo.dbc                 C   s*   t jt d}d|| f }tj||dS )Nr   zsqlite:///%s/%s)urlfuture)r   r   r   r   testing_utiltesting_engine)tempnamer@   r7   r?   r   r   r   _sqlite_file_dba   s   rD   c                 C   sT   t jt d}d| }|pdtjjjv }td||| rdnd|r%df S df S )	Nr   sqlite:///%s/foo.dbr@   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falsezsqlalchemy.future = true 	r   r   r   r   r   db	__class__
__module___write_config_file)r'   r@   r7   r?   sqlalchemy_futurer   r   r   _sqlite_testing_configg   s    %
)rO   rH   c                 C   sL   t jt d}dtjjjv }d| }td|||rdnd| r dnd|f S )Nr   r@   rE   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
    rF   rG   rI   )r'   extra_version_locationr7   rN   r?   r   r   r   _multi_dir_testing_config   s   !

rQ   
postgresqlc                 C   s"   t jt d}td|| |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   rM   )dialect
directivesr7   r   r   r   _no_sql_testing_config   s   rU   c                 C   sB   t  }t|jd}||  W d    |S 1 sw   Y  |S )Nr3   )r   r4   config_file_namer5   )textr(   r8   r   r   r   rM      s   
rM   c                  C   s<   ddl m}  tt tjstt  | tjt dS )Nr   Configztest_alembic.ini)	alembic.configrY   r   accessr   F_OKmkdirr   r   rX   r   r   r   r      s   r   asciic           
      C   s   | j |}|j}t|}|r||}t|d}|| W d    n1 s*w   Y  t	|}|r;t
| t| |}	| j |	j}|j|	jkrRtd| j j|	dd |rit||dkrddnd d S d S )Nwbz2Can't change down_revision on a refresh operation.T)_replacer   r   r   )revision_mapget_revisionr   textwrapdedentencoder4   r5   r   r:   r   r;   r   
_from_pathrevisiondown_revision	Exceptionadd_revisionr$   )
	scriptdirrev_idcontentencodingr'   oldr   fpr<   r   r   r   r   write_script  s.   



rq   c                 C   s   dd l }||  |dkr/t r/t| }t d }tj| \}}|| }t	
|| |}n|dkr9t s9t |dv s?J t| }t|tjsMJ t|  d S )Nr   r   r   )r   r   )
py_compilecompiler
   r   r:   r	   r   r   splitextr0   moveNotImplementedErrorr[   r\   r;   )r   stylerr   r<   suffixfilepathextsimple_pyc_pathr   r   r   r$     s   



r$   c                 C   s   t  }t  }t  }t| }|j|dddd t||d|  |j|dd|d t||td||dd	 |j|d
d|d t||d||f  |||fS )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 = '{}'
down_revision = '{}'

from alembic import op


def upgrade():
    op.execute("CREATE STEP 2")


def downgrade():
    op.execute("DROP STEP 2")

zutf-8)rn   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   rl   r   r%   generate_revisionrq   r   format)r(   abcr   r   r   r   three_rev_fixture7  s>   

r   c                 C   s   t  }t  }t  }t| }|j|d|ddd t||d||f  |j|d|ddd t||d||f  |j|d|ddd t||d||f  |||fS )	z:Create a multiple head fixture from the three-revs fixturezrevision d from bT)r~   splicer}   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")

zrevision 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")

zrevision 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")

)r   rl   r   r%   r   rq   )r(   r   r   r   der8   r   r   r   r   multi_heads_fixture  sF   




r   c                 C   s`   t jt d}dtjjjv }d|  }ddd | 	 D } t
d||r)dnd	|| f S )
z?alembic.ini fixture to work exactly with the 'multidb' templater   r@   z, z

c                 s   s"    | ]\}}d ||j f V  qdS )z[%s]
sqlalchemy.url = %sN)r?   ).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
    rF   rG   )r   r   r   r   r   rJ   rK   rL   keysitemsrM   )r,   r7   rN   	databasesr   r   r   _multidb_testing_config  s   
 r   )Tr   F)r>   F)FF)FrH   )rR   rH   )r^   F) r   r0   rc   r-   r   r   rH   r   rA   r   r   r   util.compatr	   r
   r   r   r*   r2   r9   r=   rD   rO   rQ   rU   rM   r   rq   r$   r   r   r   r   r   r   r   <module>   s:   
%


4
1(	
OY