sqlbag
sqlbag copied to clipboard
djrobstep
Temporary database and pytest
I'm running into a permission error on windows when using the temp file in testing. Not sure if that's me or the implementation only considers the linux environment. (https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file)
Say I have some test
@patch("builtins.input", lambda _: "yes")
def test_sync_schema():
dbh = DataBaseHandler("test", "sqlite:///:memory:")
# add a new column so we have a migration safety issue when reverting back
with dbh.engine.connect() as conn:
conn.execute('ALTER TABLE "SomeTable" ADD COLUMN "test" INTEGER')
# make sure drop column isnt just executed
with pytest.raises(UnsafeMigrationException):
sync_schema(dbh, safety=True, export=False) # < ---------------------- ERROR OCCURS HERE
# actually create it
sync_schema(dbh, safety=False, export=False)
and then the sync_schema function basically looks like the one in the documentation
def sync_schema(dbh: DataBaseHandler, safety: bool = True, export: bool = True):
if "postgresql" in dbh.url:
dialect = "postgresql"
else:
dialect = "sqlite"
with temporary_database(dialect=dialect) as temp_db_url:
# migra schema will always be kept up to date with the current object_model
migra_dbh = DataBaseHandler("migra", temp_db_url)
with dbh.session_scope() as s_current, migra_dbh.session_scope() as s_target:
m = migra.Migration(s_current, s_target)
m.set_safety(safety)
m.add_all_changes()
if m.statements:
print("THE FOLLOWING CHANGES ARE PENDING:", end="\n\n")
print(m.sql)
answer = input("Apply these changes? [yes/NO]").lower()
if answer == "yes":
# the export stuff should be irrelevant, just saving the migration statements
if export:
p = (
Path(__file__).parent
/ "schemas"
/ f"{customer}_migrate.sql"
)
if p.is_file():
p.unlink()
with open(p, "a") as file:
file.write(str(m.sql))
dbh.get_sql_schema()
print("Applying...")
m.apply()
else:
print("Not applying.")
else:
print("Already synced.")
I also tried using double nesting of temporary db (instead of the "sqlite://:memory:" in the test, but it throws the same error:
finally:
if not do_not_delete:
> os.remove(tmp.name)
E PermissionError: [WinError 32] The process cannot access the file, it's used by another process: 'C:\\Users\\~user\\AppData\\Local\\Temp\\tmp7dj3qb67'
Might also be my fault and I'm overlooking something.
Hey, thanks for filing this.
Migra itself only supports postgresql right now, and absolutely will not work on sqlite (it could be made to support it in future but would be a bunch of work).
sqlbag however supports sqlite, mysql/mariadb, and postgres, so you still should be able to create a temporary sqlite database, although I've done literally zero testing of it on windows.
What's the full exception you're seeing?
Thanks for getting back so quickly!
Ah, okay, so it might also just be due to me trying this with sqlite? Have you done some testing with the migration yourself and found a good pattern to do it with postgresql? (Especially in CI)
Concerning the full exception:
_________________________________________________ test_sync_schema _________________________________________________
x = <sqlalchemy.orm.session.Session object at 0x00000159AB7C1E08>, schema = None, exclude_schema = None
def get_inspector(x, schema=None, exclude_schema=None):
if schema and exclude_schema:
raise ValueError("Cannot provide both schema and exclude_schema")
if x is None:
return NullInspector()
c = connection_from_s_or_c(x)
try:
> ic = SUPPORTED[c.dialect.name]
E KeyError: 'sqlite'
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\schemainspect\get.py:16: KeyError
During handling of the above exception, another exception occurred:
dialect = 'sqlite', do_not_delete = False, host = ''
@contextmanager
def temporary_database(dialect="postgresql", do_not_delete=False, host=None):
"""
Args:
dialect(str): Type of database to create (either 'postgresql', 'mysql', or 'sqlite').
do_not_delete: Do not delete the database as this method usually would.
Creates a temporary database for the duration of the context manager scope. Cleans it up when finished unless do_not_delete is specified.
PostgreSQL, MySQL/MariaDB, and SQLite are supported. This method's mysql creation code uses the pymysql driver, so make sure you have that installed.
"""
host = host or ""
if dialect == "sqlite":
tmp = tempfile.NamedTemporaryFile(delete=False)
try:
url = "sqlite:///" + tmp.name
> yield url
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\sqlbag\createdrop.py:174:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
dbh = <dsDataBase.database.DataBaseHandler object at 0x00000159AB765248>, safety = True, export = False
def sync_schema(dbh: DataBaseHandler, safety: bool = True, export: bool = True):
if "postgresql" in dbh.url:
dialect = "postgresql"
else:
dialect = "sqlite"
with temporary_database(dialect=dialect) as temp_db_url:
migra_dbh = DataBaseHandler("migra", temp_db_url)
with dbh.session_scope.begin() as s_current, migra_dbh.session_scope.begin() as s_target: # pylint: disable=no-member
> m = migra.Migration(s_current, s_target)
src\dsDataBase\database.py:237:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <migra.migra.Migration object at 0x00000159AB723B08>
x_from = <sqlalchemy.orm.session.Session object at 0x00000159AB7C1E08>
x_target = <sqlalchemy.orm.session.Session object at 0x00000159AB723A08>, schema = None, exclude_schema = None
def __init__(self, x_from, x_target, schema=None, exclude_schema=None):
self.statements = Statements()
self.changes = Changes(None, None)
if schema and exclude_schema:
raise ValueError("You cannot have both a schema and excluded schema")
self.schema = schema
self.exclude_schema = exclude_schema
if isinstance(x_from, DBInspector):
self.changes.i_from = x_from
else:
self.changes.i_from = get_inspector(
> x_from, schema=schema, exclude_schema=exclude_schema
)
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\migra\migra.py:26:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
x = <sqlalchemy.orm.session.Session object at 0x00000159AB7C1E08>, schema = None, exclude_schema = None
def get_inspector(x, schema=None, exclude_schema=None):
if schema and exclude_schema:
raise ValueError("Cannot provide both schema and exclude_schema")
if x is None:
return NullInspector()
c = connection_from_s_or_c(x)
try:
ic = SUPPORTED[c.dialect.name]
except KeyError:
> raise NotImplementedError
E NotImplementedError
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\schemainspect\get.py:18: NotImplementedError
During handling of the above exception, another exception occurred:
@patch("builtins.input", lambda _: "yes")
def test_sync_schema():
dbh = DataBaseHandler("test", "sqlite:///:memory:")
with dbh.engine.connect() as conn:
conn.execute('ALTER TABLE "Plant" ADD COLUMN "test" INTEGER')
# make sure drop column isnt just executed
with pytest.raises(UnsafeMigrationException):
> sync_schema(dbh, safety=True, export=False)
src\test\test_database.py:101:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src\dsDataBase\database.py:237: in sync_schema
m = migra.Migration(s_current, s_target)
..\..\AppData\Local\Programs\Python\Python37\Lib\contextlib.py:130: in __exit__
self.gen.throw(type, value, traceback)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
dialect = 'sqlite', do_not_delete = False, host = ''
@contextmanager
def temporary_database(dialect="postgresql", do_not_delete=False, host=None):
"""
Args:
dialect(str): Type of database to create (either 'postgresql', 'mysql', or 'sqlite').
do_not_delete: Do not delete the database as this method usually would.
Creates a temporary database for the duration of the context manager scope. Cleans it up when finished unless do_not_delete is specified.
PostgreSQL, MySQL/MariaDB, and SQLite are supported. This method's mysql creation code uses the pymysql driver, so make sure you have that installed.
"""
host = host or ""
if dialect == "sqlite":
tmp = tempfile.NamedTemporaryFile(delete=False)
try:
url = "sqlite:///" + tmp.name
yield url
finally:
if not do_not_delete:
> os.remove(tmp.name)
E PermissionError: [WinError 32] Der Prozess kann nicht auf die Datei zugreifen, da sie von einem anderen Prozess verwendet wird: 'C:\\Users\\<myuser>\\AppData\\Local\\Temp\\tmpye7tg9fz'
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\sqlbag\createdrop.py:178: PermissionError
Just to add to this, if I'm using "postgresql" (locally) instead of "sqlite", I receive another error (for all tests):
kwasync = {}
if 'async' in kwargs:
kwasync['async'] = kwargs.pop('async')
if 'async_' in kwargs:
kwasync['async_'] = kwargs.pop('async_')
if dsn is None and not kwargs:
raise TypeError('missing dsn and no parameters')
dsn = _ext.make_dsn(dsn, **kwargs)
> conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
E sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) fe_sendauth: no password supplied
E
E (Background on this error at: http://sqlalche.me/e/14/e3q8)
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\psycopg2\__init__.py:127: OperationalError
Error logs are super long, but basically this occurs when
@contextmanager
def temporary_database(dialect="postgresql", do_not_delete=False, host=None):
"""
Args:
dialect(str): Type of database to create (either 'postgresql', 'mysql', or 'sqlite').
do_not_delete: Do not delete the database as this method usually would.
Creates a temporary database for the duration of the context manager scope. Cleans it up when finished unless do_not_delete is specified.
PostgreSQL, MySQL/MariaDB, and SQLite are supported. This method's mysql creation code uses the pymysql driver, so make sure you have that installed.
"""
host = host or ""
if dialect == "sqlite":
tmp = tempfile.NamedTemporaryFile(delete=False)
try:
url = "sqlite:///" + tmp.name
yield url
finally:
if not do_not_delete:
os.remove(tmp.name)
else:
tempname = temporary_name()
current_username = _current_username()
url = "{}://{}@{}/{}".format(dialect, current_username, host, tempname)
if url.startswith("mysql:"):
url = url.replace("mysql:", "mysql+pymysql:", 1)
try:
> create_database(url)
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\sqlbag\createdrop.py:191:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
db_url = 'postgresql://myuser@/sqlbag_tmp_orjuajdipi', template = None, wipe_if_existing = False
def create_database(db_url, template=None, wipe_if_existing=False):
target_url = make_url(db_url)
dbtype = target_url.get_dialect().name
if wipe_if_existing:
drop_database(db_url)
> if database_exists(target_url):
..\..\appdata\local\pypoetry\cache\virtualenvs\dsdatabase-lpbksmnq-py3.7\lib\site-packages\sqlbag\createdrop.py:85:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
db_url = postgresql://myuser@/sqlbag_tmp_orjuajdipi, test_can_select = False
def database_exists(db_url, test_can_select=False):
url = make_url(db_url)
name = url.database
db_type = url.get_dialect().name
if not test_can_select:
if db_type == "sqlite":
return name is None or name == ":memory:" or os.path.exists(name)
elif db_type in ["postgresql", "mysql"]:
> with admin_db_connection(url) as s:
I'd say because it only provides a username user@ without any password in the URL.
To fix this you need to set up that user to be able to connect to postgres without a password.
As for more general advice, there's one or two very general examples covered in the docs, but migra is definitely a roll-your-own kind of tool.
My typical workflow with it is:
- Autogenerate a migration
- Edit of necessary, check/test for correctness
- Apply changes to test environments first if you have them, check for correctness once more
- Apply changes to production, use migra again to confirm production schema now matches what you intended.
Ok, thanks. I'll follow the update schema, I also plan to use it with a double confirmation step. :)
Concerning the user setup, is there a programmatic way to do that? Or do I need to work with the config file of each individual database? How does that work for the temp dbs?
Update: I changed the user options in the pg_hba.conf to trust, and it now does not complain explicitly about the password but instead raises an OperationalError without any description.