This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
devicehub-teal/ereuse_devicehub/db.py

72 lines
2.6 KiB
Python
Raw Normal View History

2019-01-21 15:08:55 +00:00
import citext
from sqlalchemy import event
2018-08-08 19:25:53 +00:00
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import expression
from sqlalchemy_utils import view
from teal.db import SchemaSQLAlchemy, SchemaSession
class DhSession(SchemaSession):
def final_flush(self):
"""A regular flush that performs expensive final operations
through Devicehub (like saving searches), so it is thought
to be used once in each request, at the very end before
a commit.
"""
# This was done before with an ``before_commit`` sqlalchemy event
# however it is too fragile it does not detect previously-flushed
# things
# This solution makes this more aware to the user, although
# has the same problem. This is not final solution.
# todo a solution would be for this session to save, on every
# flush, all the new / dirty interesting things in a variable
# until DeviceSearch is executed
from ereuse_devicehub.resources.device.search import DeviceSearch
DeviceSearch.update_modified_devices(session=self)
2018-08-08 19:25:53 +00:00
2018-07-14 14:41:22 +00:00
2018-09-07 10:35:32 +00:00
class SQLAlchemy(SchemaSQLAlchemy):
2018-08-08 19:25:53 +00:00
"""
Superuser must create the required extensions in the public
schema of the database, as it is in the `search_path`
defined in teal.
"""
2019-01-21 15:08:55 +00:00
# todo add here all types of columns used so we don't have to
# manually import them all the time
2018-08-08 19:25:53 +00:00
UUID = postgresql.UUID
2019-01-21 15:08:55 +00:00
CIText = citext.CIText
2019-02-13 17:59:14 +00:00
PSQL_INT_MAX = 2147483648
2018-08-08 19:25:53 +00:00
2019-02-04 12:38:46 +00:00
def drop_all(self, bind='__all__', app=None, common_schema=True):
2018-07-14 14:41:22 +00:00
"""A faster nuke-like option to drop everything."""
self.drop_schema()
2019-02-04 12:38:46 +00:00
if common_schema:
self.drop_schema(schema='common')
2018-07-14 14:41:22 +00:00
def create_session(self, options):
return sessionmaker(class_=DhSession, db=self, **options)
2018-04-10 15:06:39 +00:00
def create_view(name, selectable):
"""Creates a view.
This is an adaptation from sqlalchemy_utils.view. See
`the test on sqlalchemy-utils <https://github.com/kvesteri/
sqlalchemy-utils/blob/master/tests/test_views.py>`_ for an
example on how to use.
"""
table = view.create_table_from_selectable(name, selectable)
# We need to ensure views are created / destroyed before / after
# SchemaSQLAlchemy's listeners execute
# That is why insert=True in 'after_create'
event.listen(db.metadata, 'after_create', view.CreateView(name, selectable), insert=True)
event.listen(db.metadata, 'before_drop', view.DropView(name))
return table
2019-01-21 15:08:55 +00:00
db = SQLAlchemy(session_options={'autoflush': False})
f = db.func
exp = expression