Replace usage of LegacyEngineFacade
Switch to using oslo_db.sqlalchemy.enginefacade instead, as this is required for SQLAlchemy 2.x support. Change-Id: Ifcad28239b6907b8ca396d348cbfa54185355f68
This commit is contained in:
parent
3e35eb05ed
commit
f2c103535e
@ -32,10 +32,11 @@ def run_migrations_online(target_metadata, version_table):
|
||||
:param target_metadata: Model's metadata used for autogenerate support.
|
||||
:param version_table: Override the default version table for alembic.
|
||||
"""
|
||||
engine = db.get_engine()
|
||||
with engine.connect() as connection:
|
||||
context.configure(connection=connection,
|
||||
target_metadata=target_metadata,
|
||||
version_table=version_table)
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
with db.session_for_write() as session:
|
||||
engine = session.get_bind()
|
||||
with engine.connect() as connection:
|
||||
context.configure(connection=connection,
|
||||
target_metadata=target_metadata,
|
||||
version_table=version_table)
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
@ -13,28 +13,26 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
#
|
||||
from oslo_config import cfg
|
||||
from oslo_db.sqlalchemy import session
|
||||
import threading
|
||||
|
||||
from oslo_db.sqlalchemy import enginefacade
|
||||
|
||||
_CONTEXT = threading.local()
|
||||
_FACADE = None
|
||||
|
||||
|
||||
def _create_facade_lazily():
|
||||
global _FACADE
|
||||
if _FACADE is None:
|
||||
# FIXME(priteau): Remove autocommit=True (and ideally use of
|
||||
# LegacyEngineFacade) asap since it's not compatible with SQLAlchemy
|
||||
# 2.0.
|
||||
_FACADE = session.EngineFacade.from_config(cfg.CONF, sqlite_fk=True,
|
||||
autocommit=True)
|
||||
ctx = enginefacade.transaction_context()
|
||||
ctx.configure(sqlite_fk=True)
|
||||
_FACADE = ctx
|
||||
return _FACADE
|
||||
|
||||
|
||||
def get_engine():
|
||||
facade = _create_facade_lazily()
|
||||
return facade.get_engine()
|
||||
def session_for_read():
|
||||
return _create_facade_lazily().reader.using(_CONTEXT)
|
||||
|
||||
|
||||
def get_session(**kwargs):
|
||||
facade = _create_facade_lazily()
|
||||
return facade.get_session(**kwargs)
|
||||
def session_for_write():
|
||||
return _create_facade_lazily().writer.using(_CONTEXT)
|
||||
|
@ -30,16 +30,15 @@ def get_backend():
|
||||
class State(api.State):
|
||||
|
||||
def get_state(self, name):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.StateInfo,
|
||||
session)
|
||||
q = q.filter(models.StateInfo.name == name)
|
||||
return q.value(models.StateInfo.state)
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
models.StateInfo,
|
||||
session)
|
||||
q = q.filter(models.StateInfo.name == name)
|
||||
return q.value(models.StateInfo.state)
|
||||
|
||||
def set_state(self, name, state):
|
||||
session = db.get_session()
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.StateInfo,
|
||||
@ -55,16 +54,15 @@ class State(api.State):
|
||||
return db_state.state
|
||||
|
||||
def get_metadata(self, name):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.StateInfo,
|
||||
session)
|
||||
q.filter(models.StateInfo.name == name)
|
||||
return q.value(models.StateInfo.s_metadata)
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
models.StateInfo,
|
||||
session)
|
||||
q.filter(models.StateInfo.name == name)
|
||||
return q.value(models.StateInfo.s_metadata)
|
||||
|
||||
def set_metadata(self, name, metadata):
|
||||
session = db.get_session()
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.StateInfo,
|
||||
@ -83,20 +81,19 @@ class ModuleInfo(api.ModuleInfo):
|
||||
"""Base class for module info management."""
|
||||
|
||||
def get_priority(self, name):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.ModuleStateInfo,
|
||||
session)
|
||||
q = q.filter(models.ModuleStateInfo.name == name)
|
||||
res = q.value(models.ModuleStateInfo.priority)
|
||||
if res:
|
||||
return int(res)
|
||||
else:
|
||||
return 1
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
models.ModuleStateInfo,
|
||||
session)
|
||||
q = q.filter(models.ModuleStateInfo.name == name)
|
||||
res = q.value(models.ModuleStateInfo.priority)
|
||||
if res:
|
||||
return int(res)
|
||||
else:
|
||||
return 1
|
||||
|
||||
def set_priority(self, name, priority):
|
||||
session = db.get_session()
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ModuleStateInfo,
|
||||
@ -113,20 +110,19 @@ class ModuleInfo(api.ModuleInfo):
|
||||
return int(db_state.priority)
|
||||
|
||||
def get_state(self, name):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ModuleStateInfo,
|
||||
session)
|
||||
q = q.filter(models.ModuleStateInfo.name == name)
|
||||
res = q.value(models.ModuleStateInfo.state)
|
||||
return bool(res)
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
return None
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ModuleStateInfo,
|
||||
session)
|
||||
q = q.filter(models.ModuleStateInfo.name == name)
|
||||
res = q.value(models.ModuleStateInfo.state)
|
||||
return bool(res)
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
return None
|
||||
|
||||
def set_state(self, name, state):
|
||||
session = db.get_session()
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ModuleStateInfo,
|
||||
@ -145,20 +141,19 @@ class ServiceToCollectorMapping(object):
|
||||
"""Base class for service to collector mapping."""
|
||||
|
||||
def get_mapping(self, service):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
q = q.filter(
|
||||
models.ServiceToCollectorMapping.service == service)
|
||||
return q.one()
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchMapping(service)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
q = q.filter(
|
||||
models.ServiceToCollectorMapping.service == service)
|
||||
return q.one()
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchMapping(service)
|
||||
|
||||
def set_mapping(self, service, collector):
|
||||
session = db.get_session()
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
try:
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
@ -176,37 +171,37 @@ class ServiceToCollectorMapping(object):
|
||||
return db_mapping
|
||||
|
||||
def list_services(self, collector=None):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
if collector:
|
||||
q = q.filter(
|
||||
models.ServiceToCollectorMapping.collector == collector)
|
||||
res = q.distinct().values(
|
||||
models.ServiceToCollectorMapping.service)
|
||||
return res
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
if collector:
|
||||
q = q.filter(
|
||||
models.ServiceToCollectorMapping.collector == collector)
|
||||
res = q.distinct().values(
|
||||
models.ServiceToCollectorMapping.service)
|
||||
return res
|
||||
|
||||
def list_mappings(self, collector=None):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
if collector:
|
||||
q = q.filter(
|
||||
models.ServiceToCollectorMapping.collector == collector)
|
||||
res = q.all()
|
||||
return res
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
if collector:
|
||||
q = q.filter(
|
||||
models.ServiceToCollectorMapping.collector == collector)
|
||||
res = q.all()
|
||||
return res
|
||||
|
||||
def delete_mapping(self, service):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
q = q.filter(models.ServiceToCollectorMapping.service == service)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchMapping(service)
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.ServiceToCollectorMapping,
|
||||
session)
|
||||
q = q.filter(models.ServiceToCollectorMapping.service == service)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchMapping(service)
|
||||
|
||||
|
||||
class DBAPIManager(object):
|
||||
|
@ -29,13 +29,11 @@ class HashMapBase(models.ModelBase):
|
||||
'mysql_engine': "InnoDB"}
|
||||
fk_to_resolve = {}
|
||||
|
||||
def save(self, session=None):
|
||||
def save(self):
|
||||
from cloudkitty import db
|
||||
|
||||
if session is None:
|
||||
session = db.get_session()
|
||||
|
||||
super(HashMapBase, self).save(session=session)
|
||||
with db.session_for_write() as session:
|
||||
super(HashMapBase, self).save(session=session)
|
||||
|
||||
def as_dict(self):
|
||||
d = {}
|
||||
|
@ -34,131 +34,131 @@ class HashMap(api.HashMap):
|
||||
return migration
|
||||
|
||||
def get_service(self, name=None, uuid=None):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapService)
|
||||
if name:
|
||||
q = q.filter(
|
||||
models.HashMapService.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == uuid)
|
||||
else:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either name or uuid.')
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchService(name=name, uuid=uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapService)
|
||||
if name:
|
||||
q = q.filter(
|
||||
models.HashMapService.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == uuid)
|
||||
else:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either name or uuid.')
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchService(name=name, uuid=uuid)
|
||||
|
||||
def get_field(self, uuid=None, service_uuid=None, name=None):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapField)
|
||||
if uuid:
|
||||
q = q.filter(
|
||||
models.HashMapField.field_id == uuid)
|
||||
elif service_uuid and name:
|
||||
q = q.join(
|
||||
models.HashMapField.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid,
|
||||
models.HashMapField.name == name)
|
||||
else:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either a uuid'
|
||||
' or a service_uuid and a name.')
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchField(uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapField)
|
||||
if uuid:
|
||||
q = q.filter(
|
||||
models.HashMapField.field_id == uuid)
|
||||
elif service_uuid and name:
|
||||
q = q.join(
|
||||
models.HashMapField.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid,
|
||||
models.HashMapField.name == name)
|
||||
else:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either a uuid'
|
||||
' or a service_uuid and a name.')
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchField(uuid)
|
||||
|
||||
def get_group(self, uuid=None, name=None):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapGroup)
|
||||
if uuid:
|
||||
q = q.filter(
|
||||
models.HashMapGroup.group_id == uuid)
|
||||
if name:
|
||||
q = q.filter(
|
||||
models.HashMapGroup.name == name)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchGroup(name, uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapGroup)
|
||||
if uuid:
|
||||
q = q.filter(
|
||||
models.HashMapGroup.group_id == uuid)
|
||||
if name:
|
||||
q = q.filter(
|
||||
models.HashMapGroup.name == name)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchGroup(name, uuid)
|
||||
|
||||
def get_mapping(self, uuid):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapMapping)
|
||||
q = q.filter(
|
||||
models.HashMapMapping.mapping_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchMapping(uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapMapping)
|
||||
q = q.filter(
|
||||
models.HashMapMapping.mapping_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchMapping(uuid)
|
||||
|
||||
def get_threshold(self, uuid):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapThreshold)
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.threshold_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchThreshold(uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapThreshold)
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.threshold_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchThreshold(uuid)
|
||||
|
||||
def get_group_from_mapping(self, uuid):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapGroup)
|
||||
q = q.join(
|
||||
models.HashMapGroup.mappings)
|
||||
q = q.filter(
|
||||
models.HashMapMapping.mapping_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.MappingHasNoGroup(uuid=uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapGroup)
|
||||
q = q.join(
|
||||
models.HashMapGroup.mappings)
|
||||
q = q.filter(
|
||||
models.HashMapMapping.mapping_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.MappingHasNoGroup(uuid=uuid)
|
||||
|
||||
def get_group_from_threshold(self, uuid):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.HashMapGroup)
|
||||
q = q.join(
|
||||
models.HashMapGroup.thresholds)
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.threshold_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.ThresholdHasNoGroup(uuid=uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.HashMapGroup)
|
||||
q = q.join(
|
||||
models.HashMapGroup.thresholds)
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.threshold_id == uuid)
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.ThresholdHasNoGroup(uuid=uuid)
|
||||
|
||||
def list_services(self):
|
||||
session = db.get_session()
|
||||
q = session.query(models.HashMapService)
|
||||
res = q.values(
|
||||
models.HashMapService.service_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
with db.session_for_read() as session:
|
||||
q = session.query(models.HashMapService)
|
||||
res = q.values(
|
||||
models.HashMapService.service_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
|
||||
def list_fields(self, service_uuid):
|
||||
session = db.get_session()
|
||||
q = session.query(models.HashMapField)
|
||||
q = q.join(
|
||||
models.HashMapField.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid)
|
||||
res = q.values(models.HashMapField.field_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
with db.session_for_read() as session:
|
||||
q = session.query(models.HashMapField)
|
||||
q = q.join(
|
||||
models.HashMapField.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid)
|
||||
res = q.values(models.HashMapField.field_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
|
||||
def list_groups(self):
|
||||
session = db.get_session()
|
||||
q = session.query(models.HashMapGroup)
|
||||
res = q.values(
|
||||
models.HashMapGroup.group_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
with db.session_for_read() as session:
|
||||
q = session.query(models.HashMapGroup)
|
||||
res = q.values(
|
||||
models.HashMapGroup.group_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
|
||||
def list_mappings(self,
|
||||
service_uuid=None,
|
||||
@ -167,33 +167,34 @@ class HashMap(api.HashMap):
|
||||
no_group=False,
|
||||
**kwargs):
|
||||
|
||||
session = db.get_session()
|
||||
q = session.query(models.HashMapMapping)
|
||||
if service_uuid:
|
||||
q = q.join(
|
||||
models.HashMapMapping.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid)
|
||||
elif field_uuid:
|
||||
q = q.join(
|
||||
models.HashMapMapping.field)
|
||||
q = q.filter(models.HashMapField.field_id == field_uuid)
|
||||
elif not service_uuid and not field_uuid and not group_uuid:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either service_uuid,'
|
||||
' field_uuid or group_uuid.')
|
||||
if 'tenant_uuid' in kwargs:
|
||||
q = q.filter(
|
||||
models.HashMapMapping.tenant_id == kwargs.get('tenant_uuid'))
|
||||
if group_uuid:
|
||||
q = q.join(
|
||||
models.HashMapMapping.group)
|
||||
q = q.filter(models.HashMapGroup.group_id == group_uuid)
|
||||
elif no_group:
|
||||
q = q.filter(models.HashMapMapping.group_id == None) # noqa
|
||||
res = q.values(
|
||||
models.HashMapMapping.mapping_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
with db.session_for_read() as session:
|
||||
q = session.query(models.HashMapMapping)
|
||||
if service_uuid:
|
||||
q = q.join(
|
||||
models.HashMapMapping.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid)
|
||||
elif field_uuid:
|
||||
q = q.join(
|
||||
models.HashMapMapping.field)
|
||||
q = q.filter(models.HashMapField.field_id == field_uuid)
|
||||
elif not service_uuid and not field_uuid and not group_uuid:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either service_uuid,'
|
||||
' field_uuid or group_uuid.')
|
||||
if 'tenant_uuid' in kwargs:
|
||||
q = q.filter(
|
||||
models.HashMapMapping.tenant_id == kwargs.get(
|
||||
'tenant_uuid'))
|
||||
if group_uuid:
|
||||
q = q.join(
|
||||
models.HashMapMapping.group)
|
||||
q = q.filter(models.HashMapGroup.group_id == group_uuid)
|
||||
elif no_group:
|
||||
q = q.filter(models.HashMapMapping.group_id == None) # noqa
|
||||
res = q.values(
|
||||
models.HashMapMapping.mapping_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
|
||||
def list_thresholds(self,
|
||||
service_uuid=None,
|
||||
@ -202,38 +203,38 @@ class HashMap(api.HashMap):
|
||||
no_group=False,
|
||||
**kwargs):
|
||||
|
||||
session = db.get_session()
|
||||
q = session.query(models.HashMapThreshold)
|
||||
if service_uuid:
|
||||
q = q.join(
|
||||
models.HashMapThreshold.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid)
|
||||
elif field_uuid:
|
||||
q = q.join(
|
||||
models.HashMapThreshold.field)
|
||||
q = q.filter(models.HashMapField.field_id == field_uuid)
|
||||
elif not service_uuid and not field_uuid and not group_uuid:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either service_uuid,'
|
||||
' field_uuid or group_uuid.')
|
||||
if 'tenant_uuid' in kwargs:
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.tenant_id == kwargs.get('tenant_uuid'))
|
||||
if group_uuid:
|
||||
q = q.join(
|
||||
models.HashMapThreshold.group)
|
||||
q = q.filter(models.HashMapGroup.group_id == group_uuid)
|
||||
elif no_group:
|
||||
q = q.filter(models.HashMapThreshold.group_id == None) # noqa
|
||||
res = q.values(
|
||||
models.HashMapThreshold.threshold_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
with db.session_for_read() as session:
|
||||
q = session.query(models.HashMapThreshold)
|
||||
if service_uuid:
|
||||
q = q.join(
|
||||
models.HashMapThreshold.service)
|
||||
q = q.filter(
|
||||
models.HashMapService.service_id == service_uuid)
|
||||
elif field_uuid:
|
||||
q = q.join(
|
||||
models.HashMapThreshold.field)
|
||||
q = q.filter(models.HashMapField.field_id == field_uuid)
|
||||
elif not service_uuid and not field_uuid and not group_uuid:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either service_uuid,'
|
||||
' field_uuid or group_uuid.')
|
||||
if 'tenant_uuid' in kwargs:
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.tenant_id == kwargs.get(
|
||||
'tenant_uuid'))
|
||||
if group_uuid:
|
||||
q = q.join(
|
||||
models.HashMapThreshold.group)
|
||||
q = q.filter(models.HashMapGroup.group_id == group_uuid)
|
||||
elif no_group:
|
||||
q = q.filter(models.HashMapThreshold.group_id == None) # noqa
|
||||
res = q.values(
|
||||
models.HashMapThreshold.threshold_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
|
||||
def create_service(self, name):
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
service_db = models.HashMapService(name=name)
|
||||
service_db.service_id = uuidutils.generate_uuid()
|
||||
session.add(service_db)
|
||||
@ -246,9 +247,8 @@ class HashMap(api.HashMap):
|
||||
|
||||
def create_field(self, service_uuid, name):
|
||||
service_db = self.get_service(uuid=service_uuid)
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
field_db = models.HashMapField(
|
||||
service_id=service_db.id,
|
||||
name=name,
|
||||
@ -264,9 +264,8 @@ class HashMap(api.HashMap):
|
||||
return field_db
|
||||
|
||||
def create_group(self, name):
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
group_db = models.HashMapGroup(
|
||||
name=name,
|
||||
group_id=uuidutils.generate_uuid())
|
||||
@ -308,9 +307,8 @@ class HashMap(api.HashMap):
|
||||
if group_id:
|
||||
group_db = self.get_group(uuid=group_id)
|
||||
group_fk = group_db.id
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
field_map = models.HashMapMapping(
|
||||
mapping_id=uuidutils.generate_uuid(),
|
||||
value=value,
|
||||
@ -365,9 +363,8 @@ class HashMap(api.HashMap):
|
||||
if group_id:
|
||||
group_db = self.get_group(uuid=group_id)
|
||||
group_fk = group_db.id
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
threshold_db = models.HashMapThreshold(
|
||||
threshold_id=uuidutils.generate_uuid(),
|
||||
level=level,
|
||||
@ -395,9 +392,8 @@ class HashMap(api.HashMap):
|
||||
return threshold_db
|
||||
|
||||
def update_mapping(self, uuid, **kwargs):
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
q = session.query(models.HashMapMapping)
|
||||
q = q.filter(
|
||||
models.HashMapMapping.mapping_id == uuid)
|
||||
@ -442,9 +438,8 @@ class HashMap(api.HashMap):
|
||||
raise api.NoSuchMapping(uuid)
|
||||
|
||||
def update_threshold(self, uuid, **kwargs):
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
q = session.query(models.HashMapThreshold)
|
||||
q = q.filter(
|
||||
models.HashMapThreshold.threshold_id == uuid)
|
||||
@ -483,38 +478,37 @@ class HashMap(api.HashMap):
|
||||
raise api.NoSuchThreshold(uuid)
|
||||
|
||||
def delete_service(self, name=None, uuid=None):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.HashMapService,
|
||||
session)
|
||||
if name:
|
||||
q = q.filter(models.HashMapService.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(models.HashMapService.service_id == uuid)
|
||||
else:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either name or uuid.')
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchService(name, uuid)
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.HashMapService,
|
||||
session)
|
||||
if name:
|
||||
q = q.filter(models.HashMapService.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(models.HashMapService.service_id == uuid)
|
||||
else:
|
||||
raise api.ClientHashMapError(
|
||||
'You must specify either name or uuid.')
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchService(name, uuid)
|
||||
|
||||
def delete_field(self, uuid):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.HashMapField,
|
||||
session)
|
||||
q = q.filter(models.HashMapField.field_id == uuid)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchField(uuid)
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.HashMapField,
|
||||
session)
|
||||
q = q.filter(models.HashMapField.field_id == uuid)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchField(uuid)
|
||||
|
||||
def delete_group(self, uuid, recurse=True):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.HashMapGroup,
|
||||
session)
|
||||
q = q.filter(models.HashMapGroup.group_id == uuid)
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.HashMapGroup,
|
||||
session)
|
||||
q = q.filter(models.HashMapGroup.group_id == uuid)
|
||||
try:
|
||||
r = q.with_for_update().one()
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
@ -527,21 +521,21 @@ class HashMap(api.HashMap):
|
||||
q.delete()
|
||||
|
||||
def delete_mapping(self, uuid):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.HashMapMapping,
|
||||
session)
|
||||
q = q.filter(models.HashMapMapping.mapping_id == uuid)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchMapping(uuid)
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.HashMapMapping,
|
||||
session)
|
||||
q = q.filter(models.HashMapMapping.mapping_id == uuid)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchMapping(uuid)
|
||||
|
||||
def delete_threshold(self, uuid):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.HashMapThreshold,
|
||||
session)
|
||||
q = q.filter(models.HashMapThreshold.threshold_id == uuid)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchThreshold(uuid)
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.HashMapThreshold,
|
||||
session)
|
||||
q = q.filter(models.HashMapThreshold.threshold_id == uuid)
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchThreshold(uuid)
|
||||
|
@ -29,13 +29,11 @@ class HashMapBase(models.ModelBase):
|
||||
'mysql_engine': "InnoDB"}
|
||||
fk_to_resolve = {}
|
||||
|
||||
def save(self, session=None):
|
||||
def save(self):
|
||||
from cloudkitty import db
|
||||
|
||||
if session is None:
|
||||
session = db.get_session()
|
||||
|
||||
super(HashMapBase, self).save(session=session)
|
||||
with db.session_for_write() as session:
|
||||
super(HashMapBase, self).save(session=session)
|
||||
|
||||
def as_dict(self):
|
||||
d = {}
|
||||
|
@ -34,33 +34,32 @@ class PyScripts(api.PyScripts):
|
||||
return migration
|
||||
|
||||
def get_script(self, name=None, uuid=None):
|
||||
session = db.get_session()
|
||||
try:
|
||||
q = session.query(models.PyScriptsScript)
|
||||
if name:
|
||||
q = q.filter(
|
||||
models.PyScriptsScript.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(
|
||||
models.PyScriptsScript.script_id == uuid)
|
||||
else:
|
||||
raise ValueError('You must specify either name or uuid.')
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchScript(name=name, uuid=uuid)
|
||||
with db.session_for_read() as session:
|
||||
try:
|
||||
q = session.query(models.PyScriptsScript)
|
||||
if name:
|
||||
q = q.filter(
|
||||
models.PyScriptsScript.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(
|
||||
models.PyScriptsScript.script_id == uuid)
|
||||
else:
|
||||
raise ValueError('You must specify either name or uuid.')
|
||||
res = q.one()
|
||||
return res
|
||||
except sqlalchemy.orm.exc.NoResultFound:
|
||||
raise api.NoSuchScript(name=name, uuid=uuid)
|
||||
|
||||
def list_scripts(self):
|
||||
session = db.get_session()
|
||||
q = session.query(models.PyScriptsScript)
|
||||
res = q.values(
|
||||
models.PyScriptsScript.script_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
with db.session_for_read() as session:
|
||||
q = session.query(models.PyScriptsScript)
|
||||
res = q.values(
|
||||
models.PyScriptsScript.script_id)
|
||||
return [uuid[0] for uuid in res]
|
||||
|
||||
def create_script(self, name, data):
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
script_db = models.PyScriptsScript(name=name)
|
||||
script_db.data = data
|
||||
script_db.script_id = uuidutils.generate_uuid()
|
||||
@ -73,9 +72,8 @@ class PyScripts(api.PyScripts):
|
||||
script_db.script_id)
|
||||
|
||||
def update_script(self, uuid, **kwargs):
|
||||
session = db.get_session()
|
||||
try:
|
||||
with session.begin():
|
||||
with db.session_for_write() as session:
|
||||
q = session.query(models.PyScriptsScript)
|
||||
q = q.filter(
|
||||
models.PyScriptsScript.script_id == uuid
|
||||
@ -99,16 +97,16 @@ class PyScripts(api.PyScripts):
|
||||
raise api.NoSuchScript(uuid=uuid)
|
||||
|
||||
def delete_script(self, name=None, uuid=None):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
models.PyScriptsScript,
|
||||
session)
|
||||
if name:
|
||||
q = q.filter(models.PyScriptsScript.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(models.PyScriptsScript.script_id == uuid)
|
||||
else:
|
||||
raise ValueError('You must specify either name or uuid.')
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchScript(uuid=uuid)
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
models.PyScriptsScript,
|
||||
session)
|
||||
if name:
|
||||
q = q.filter(models.PyScriptsScript.name == name)
|
||||
elif uuid:
|
||||
q = q.filter(models.PyScriptsScript.script_id == uuid)
|
||||
else:
|
||||
raise ValueError('You must specify either name or uuid.')
|
||||
r = q.delete()
|
||||
if not r:
|
||||
raise api.NoSuchScript(uuid=uuid)
|
||||
|
@ -29,13 +29,11 @@ class PyScriptsBase(models.ModelBase):
|
||||
'mysql_engine': "InnoDB"}
|
||||
fk_to_resolve = {}
|
||||
|
||||
def save(self, session=None):
|
||||
def save(self):
|
||||
from cloudkitty import db
|
||||
|
||||
if session is None:
|
||||
session = db.get_session()
|
||||
|
||||
super(PyScriptsBase, self).save(session=session)
|
||||
with db.session_for_write() as session:
|
||||
super(PyScriptsBase, self).save(session=session)
|
||||
|
||||
def as_dict(self):
|
||||
d = {}
|
||||
|
@ -56,45 +56,37 @@ class HybridStorage(BaseStorage):
|
||||
HYBRID_BACKENDS_NAMESPACE,
|
||||
cfg.CONF.storage_hybrid.backend,
|
||||
invoke_on_load=True).driver
|
||||
self._sql_session = {}
|
||||
|
||||
def _check_session(self, tenant_id):
|
||||
session = self._sql_session.get(tenant_id, None)
|
||||
if not session:
|
||||
self._sql_session[tenant_id] = db.get_session()
|
||||
self._sql_session[tenant_id].begin()
|
||||
|
||||
def init(self):
|
||||
migration.upgrade('head')
|
||||
self._hybrid_backend.init()
|
||||
|
||||
def get_state(self, tenant_id=None):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(self.state_model, session)
|
||||
if tenant_id:
|
||||
q = q.filter(self.state_model.tenant_id == tenant_id)
|
||||
q = q.order_by(self.state_model.state.desc())
|
||||
r = q.first()
|
||||
return r.state if r else None
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(self.state_model, session)
|
||||
if tenant_id:
|
||||
q = q.filter(self.state_model.tenant_id == tenant_id)
|
||||
q = q.order_by(self.state_model.state.desc())
|
||||
r = q.first()
|
||||
return r.state if r else None
|
||||
|
||||
def _set_state(self, tenant_id, state):
|
||||
self._check_session(tenant_id)
|
||||
session = self._sql_session[tenant_id]
|
||||
q = utils.model_query(self.state_model, session)
|
||||
if tenant_id:
|
||||
q = q.filter(self.state_model.tenant_id == tenant_id)
|
||||
r = q.first()
|
||||
do_commit = False
|
||||
if r:
|
||||
if state > r.state:
|
||||
q.update({'state': state})
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(self.state_model, session)
|
||||
if tenant_id:
|
||||
q = q.filter(self.state_model.tenant_id == tenant_id)
|
||||
r = q.first()
|
||||
do_commit = False
|
||||
if r:
|
||||
if state > r.state:
|
||||
q.update({'state': state})
|
||||
do_commit = True
|
||||
else:
|
||||
state = self.state_model(tenant_id=tenant_id, state=state)
|
||||
session.add(state)
|
||||
do_commit = True
|
||||
else:
|
||||
state = self.state_model(tenant_id=tenant_id, state=state)
|
||||
session.add(state)
|
||||
do_commit = True
|
||||
if do_commit:
|
||||
session.commit()
|
||||
if do_commit:
|
||||
session.commit()
|
||||
|
||||
def _commit(self, tenant_id):
|
||||
self._hybrid_backend.commit(tenant_id, self.get_state(tenant_id))
|
||||
@ -105,7 +97,6 @@ class HybridStorage(BaseStorage):
|
||||
def _post_commit(self, tenant_id):
|
||||
self._set_state(tenant_id, self.usage_start_dt.get(tenant_id))
|
||||
super(HybridStorage, self)._post_commit(tenant_id)
|
||||
del self._sql_session[tenant_id]
|
||||
|
||||
def get_total(self, begin=None, end=None, tenant_id=None,
|
||||
service=None, groupby=None):
|
||||
|
@ -35,140 +35,131 @@ class SQLAlchemyStorage(storage.BaseStorage):
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super(SQLAlchemyStorage, self).__init__(**kwargs)
|
||||
self._session = {}
|
||||
|
||||
@staticmethod
|
||||
def init():
|
||||
migration.upgrade('head')
|
||||
|
||||
def _pre_commit(self, tenant_id):
|
||||
self._check_session(tenant_id)
|
||||
if not self._has_data.get(tenant_id):
|
||||
empty_frame = {'vol': {'qty': 0, 'unit': 'None'},
|
||||
'rating': {'price': 0}, 'desc': ''}
|
||||
self._append_time_frame('_NO_DATA_', empty_frame, tenant_id)
|
||||
|
||||
def _commit(self, tenant_id):
|
||||
self._session[tenant_id].commit()
|
||||
super(SQLAlchemyStorage, self)._commit(tenant_id)
|
||||
|
||||
def _post_commit(self, tenant_id):
|
||||
super(SQLAlchemyStorage, self)._post_commit(tenant_id)
|
||||
del self._session[tenant_id]
|
||||
|
||||
def _check_session(self, tenant_id):
|
||||
session = self._session.get(tenant_id)
|
||||
if not session:
|
||||
self._session[tenant_id] = db.get_session()
|
||||
self._session[tenant_id].begin()
|
||||
|
||||
def _dispatch(self, data, tenant_id):
|
||||
self._check_session(tenant_id)
|
||||
for service in data:
|
||||
for frame in data[service]:
|
||||
self._append_time_frame(service, frame, tenant_id)
|
||||
self._has_data[tenant_id] = True
|
||||
|
||||
def get_state(self, tenant_id=None):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
self.frame_model,
|
||||
session)
|
||||
if tenant_id:
|
||||
q = q.filter(
|
||||
self.frame_model.tenant_id == tenant_id)
|
||||
q = q.order_by(
|
||||
self.frame_model.begin.desc())
|
||||
r = q.first()
|
||||
if r:
|
||||
return r.begin
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
self.frame_model,
|
||||
session)
|
||||
if tenant_id:
|
||||
q = q.filter(
|
||||
self.frame_model.tenant_id == tenant_id)
|
||||
q = q.order_by(
|
||||
self.frame_model.begin.desc())
|
||||
r = q.first()
|
||||
if r:
|
||||
return r.begin
|
||||
|
||||
def get_total(self, begin=None, end=None, tenant_id=None, service=None,
|
||||
groupby=None):
|
||||
session = db.get_session()
|
||||
querymodels = [
|
||||
sqlalchemy.func.sum(self.frame_model.rate).label('rate')
|
||||
]
|
||||
with db.session_for_read() as session:
|
||||
querymodels = [
|
||||
sqlalchemy.func.sum(self.frame_model.rate).label('rate')
|
||||
]
|
||||
|
||||
if not begin:
|
||||
begin = ck_utils.get_month_start_timestamp()
|
||||
if not end:
|
||||
end = ck_utils.get_next_month_timestamp()
|
||||
# Boundary calculation
|
||||
if tenant_id:
|
||||
querymodels.append(self.frame_model.tenant_id)
|
||||
if service:
|
||||
querymodels.append(self.frame_model.res_type)
|
||||
if groupby:
|
||||
groupbyfields = groupby.split(",")
|
||||
for field in groupbyfields:
|
||||
field_obj = self.frame_model.__dict__.get(field, None)
|
||||
if field_obj and field_obj not in querymodels:
|
||||
querymodels.append(field_obj)
|
||||
if not begin:
|
||||
begin = ck_utils.get_month_start_timestamp()
|
||||
if not end:
|
||||
end = ck_utils.get_next_month_timestamp()
|
||||
# Boundary calculation
|
||||
if tenant_id:
|
||||
querymodels.append(self.frame_model.tenant_id)
|
||||
if service:
|
||||
querymodels.append(self.frame_model.res_type)
|
||||
if groupby:
|
||||
groupbyfields = groupby.split(",")
|
||||
for field in groupbyfields:
|
||||
field_obj = self.frame_model.__dict__.get(field, None)
|
||||
if field_obj and field_obj not in querymodels:
|
||||
querymodels.append(field_obj)
|
||||
|
||||
q = session.query(*querymodels)
|
||||
if tenant_id:
|
||||
q = session.query(*querymodels)
|
||||
if tenant_id:
|
||||
q = q.filter(
|
||||
self.frame_model.tenant_id == tenant_id)
|
||||
if service:
|
||||
q = q.filter(
|
||||
self.frame_model.res_type == service)
|
||||
# begin and end filters are both needed, do not remove one of them.
|
||||
q = q.filter(
|
||||
self.frame_model.tenant_id == tenant_id)
|
||||
if service:
|
||||
q = q.filter(
|
||||
self.frame_model.res_type == service)
|
||||
# begin and end filters are both needed, do not remove one of them.
|
||||
q = q.filter(
|
||||
self.frame_model.begin.between(begin, end),
|
||||
self.frame_model.end.between(begin, end),
|
||||
self.frame_model.res_type != '_NO_DATA_')
|
||||
if groupby:
|
||||
q = q.group_by(sqlalchemy.sql.text(groupby))
|
||||
self.frame_model.begin.between(begin, end),
|
||||
self.frame_model.end.between(begin, end),
|
||||
self.frame_model.res_type != '_NO_DATA_')
|
||||
if groupby:
|
||||
q = q.group_by(sqlalchemy.sql.text(groupby))
|
||||
|
||||
# Order by sum(rate)
|
||||
q = q.order_by(sqlalchemy.func.sum(self.frame_model.rate))
|
||||
results = q.all()
|
||||
totallist = []
|
||||
for r in results:
|
||||
total = {model.name: value for model, value in zip(querymodels, r)}
|
||||
total["begin"] = begin
|
||||
total["end"] = end
|
||||
totallist.append(total)
|
||||
# Order by sum(rate)
|
||||
q = q.order_by(sqlalchemy.func.sum(self.frame_model.rate))
|
||||
results = q.all()
|
||||
totallist = []
|
||||
for r in results:
|
||||
total = {model.name: value for model, value in zip(querymodels,
|
||||
r)}
|
||||
total["begin"] = begin
|
||||
total["end"] = end
|
||||
totallist.append(total)
|
||||
|
||||
return totallist
|
||||
return totallist
|
||||
|
||||
def get_tenants(self, begin, end):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
self.frame_model,
|
||||
session)
|
||||
# begin and end filters are both needed, do not remove one of them.
|
||||
q = q.filter(
|
||||
self.frame_model.begin.between(begin, end),
|
||||
self.frame_model.end.between(begin, end))
|
||||
tenants = q.distinct().values(
|
||||
self.frame_model.tenant_id)
|
||||
return [tenant.tenant_id for tenant in tenants]
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
self.frame_model,
|
||||
session)
|
||||
# begin and end filters are both needed, do not remove one of them.
|
||||
q = q.filter(
|
||||
self.frame_model.begin.between(begin, end),
|
||||
self.frame_model.end.between(begin, end))
|
||||
tenants = q.distinct().values(
|
||||
self.frame_model.tenant_id)
|
||||
return [tenant.tenant_id for tenant in tenants]
|
||||
|
||||
def get_time_frame(self, begin, end, **filters):
|
||||
if not begin:
|
||||
begin = ck_utils.get_month_start()
|
||||
if not end:
|
||||
end = ck_utils.get_next_month()
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
self.frame_model,
|
||||
session)
|
||||
# begin and end filters are both needed, do not remove one of them.
|
||||
q = q.filter(
|
||||
self.frame_model.begin.between(begin, end),
|
||||
self.frame_model.end.between(begin, end))
|
||||
for filter_name, filter_value in filters.items():
|
||||
if filter_value:
|
||||
q = q.filter(
|
||||
getattr(self.frame_model, filter_name) == filter_value)
|
||||
if not filters.get('res_type'):
|
||||
q = q.filter(self.frame_model.res_type != '_NO_DATA_')
|
||||
count = q.count()
|
||||
if not count:
|
||||
raise NoTimeFrame()
|
||||
r = q.all()
|
||||
return [entry.to_cloudkitty(self._collector) for entry in r]
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(
|
||||
self.frame_model,
|
||||
session)
|
||||
# begin and end filters are both needed, do not remove one of them.
|
||||
q = q.filter(
|
||||
self.frame_model.begin.between(begin, end),
|
||||
self.frame_model.end.between(begin, end))
|
||||
for filter_name, filter_value in filters.items():
|
||||
if filter_value:
|
||||
q = q.filter(
|
||||
getattr(self.frame_model, filter_name) == filter_value)
|
||||
if not filters.get('res_type'):
|
||||
q = q.filter(self.frame_model.res_type != '_NO_DATA_')
|
||||
count = q.count()
|
||||
if not count:
|
||||
raise NoTimeFrame()
|
||||
r = q.all()
|
||||
return [entry.to_cloudkitty(self._collector) for entry in r]
|
||||
|
||||
def _append_time_frame(self, res_type, frame, tenant_id):
|
||||
vol_dict = frame['vol']
|
||||
@ -201,4 +192,5 @@ class SQLAlchemyStorage(storage.BaseStorage):
|
||||
:param desc: Resource description (metadata).
|
||||
"""
|
||||
frame = self.frame_model(**kwargs)
|
||||
self._session[kwargs.get('tenant_id')].add(frame)
|
||||
with db.session_for_write() as session:
|
||||
session.add(frame)
|
||||
|
@ -85,28 +85,26 @@ class StateManager(object):
|
||||
:param offset: optional to shift the projection
|
||||
:type offset: int
|
||||
"""
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
with db.session_for_read() as session:
|
||||
|
||||
q = utils.model_query(self.model, session)
|
||||
if identifier:
|
||||
q = q.filter(
|
||||
self.model.identifier.in_(to_list_if_needed(identifier)))
|
||||
if fetcher:
|
||||
q = q.filter(
|
||||
self.model.fetcher.in_(to_list_if_needed(fetcher)))
|
||||
if collector:
|
||||
q = q.filter(
|
||||
self.model.collector.in_(to_list_if_needed(collector)))
|
||||
if scope_key:
|
||||
q = q.filter(
|
||||
self.model.scope_key.in_(to_list_if_needed(scope_key)))
|
||||
if active is not None and active != []:
|
||||
q = q.filter(self.model.active.in_(to_list_if_needed(active)))
|
||||
q = apply_offset_and_limit(limit, offset, q)
|
||||
q = utils.model_query(self.model, session)
|
||||
if identifier:
|
||||
q = q.filter(
|
||||
self.model.identifier.in_(to_list_if_needed(identifier)))
|
||||
if fetcher:
|
||||
q = q.filter(
|
||||
self.model.fetcher.in_(to_list_if_needed(fetcher)))
|
||||
if collector:
|
||||
q = q.filter(
|
||||
self.model.collector.in_(to_list_if_needed(collector)))
|
||||
if scope_key:
|
||||
q = q.filter(
|
||||
self.model.scope_key.in_(to_list_if_needed(scope_key)))
|
||||
if active is not None and active != []:
|
||||
q = q.filter(self.model.active.in_(to_list_if_needed(active)))
|
||||
q = apply_offset_and_limit(limit, offset, q)
|
||||
|
||||
r = q.all()
|
||||
session.close()
|
||||
r = q.all()
|
||||
|
||||
for item in r:
|
||||
item.last_processed_timestamp = tzutils.utc_to_local(
|
||||
@ -183,20 +181,18 @@ class StateManager(object):
|
||||
"""
|
||||
last_processed_timestamp = tzutils.local_to_utc(
|
||||
last_processed_timestamp, naive=True)
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
r = self._get_db_item(
|
||||
session, identifier, fetcher, collector, scope_key)
|
||||
with db.session_for_write() as session:
|
||||
r = self._get_db_item(
|
||||
session, identifier, fetcher, collector, scope_key)
|
||||
|
||||
if r:
|
||||
if r.last_processed_timestamp != last_processed_timestamp:
|
||||
r.last_processed_timestamp = last_processed_timestamp
|
||||
session.commit()
|
||||
else:
|
||||
self.create_scope(identifier, last_processed_timestamp,
|
||||
fetcher=fetcher, collector=collector,
|
||||
scope_key=scope_key)
|
||||
session.close()
|
||||
if r:
|
||||
if r.last_processed_timestamp != last_processed_timestamp:
|
||||
r.last_processed_timestamp = last_processed_timestamp
|
||||
session.commit()
|
||||
else:
|
||||
self.create_scope(identifier, last_processed_timestamp,
|
||||
fetcher=fetcher, collector=collector,
|
||||
scope_key=scope_key)
|
||||
|
||||
def create_scope(self, identifier, last_processed_timestamp, fetcher=None,
|
||||
collector=None, scope_key=None, active=True,
|
||||
@ -219,25 +215,18 @@ class StateManager(object):
|
||||
:type session: object
|
||||
"""
|
||||
|
||||
is_session_reused = True
|
||||
if not session:
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
is_session_reused = False
|
||||
with db.session_for_write() as session:
|
||||
|
||||
state_object = self.model(
|
||||
identifier=identifier,
|
||||
last_processed_timestamp=last_processed_timestamp,
|
||||
fetcher=fetcher,
|
||||
collector=collector,
|
||||
scope_key=scope_key,
|
||||
active=active
|
||||
)
|
||||
session.add(state_object)
|
||||
session.commit()
|
||||
|
||||
if not is_session_reused:
|
||||
session.close()
|
||||
state_object = self.model(
|
||||
identifier=identifier,
|
||||
last_processed_timestamp=last_processed_timestamp,
|
||||
fetcher=fetcher,
|
||||
collector=collector,
|
||||
scope_key=scope_key,
|
||||
active=active
|
||||
)
|
||||
session.add(state_object)
|
||||
session.commit()
|
||||
|
||||
def get_state(self, identifier,
|
||||
fetcher=None, collector=None, scope_key=None):
|
||||
@ -261,11 +250,9 @@ class StateManager(object):
|
||||
:type scope_key: str
|
||||
:rtype: datetime.datetime
|
||||
"""
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
r = self._get_db_item(
|
||||
session, identifier, fetcher, collector, scope_key)
|
||||
session.close()
|
||||
with db.session_for_read() as session:
|
||||
r = self._get_db_item(
|
||||
session, identifier, fetcher, collector, scope_key)
|
||||
return tzutils.utc_to_local(r.last_processed_timestamp) if r else None
|
||||
|
||||
def init(self):
|
||||
@ -274,10 +261,8 @@ class StateManager(object):
|
||||
# This is made in order to stay compatible with legacy behavior but
|
||||
# shouldn't be used
|
||||
def get_tenants(self, begin=None, end=None):
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
q = utils.model_query(self.model, session)
|
||||
session.close()
|
||||
with db.session_for_read() as session:
|
||||
q = utils.model_query(self.model, session)
|
||||
return [tenant.identifier for tenant in q]
|
||||
|
||||
def update_storage_scope(self, storage_scope_to_update, scope_key=None,
|
||||
@ -295,30 +280,28 @@ class StateManager(object):
|
||||
:param active: indicates if the storage scope is active for processing
|
||||
:type active: bool
|
||||
"""
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
with db.session_for_write() as session:
|
||||
|
||||
db_scope = self._get_db_item(session,
|
||||
storage_scope_to_update.identifier,
|
||||
storage_scope_to_update.fetcher,
|
||||
storage_scope_to_update.collector,
|
||||
storage_scope_to_update.scope_key)
|
||||
db_scope = self._get_db_item(session,
|
||||
storage_scope_to_update.identifier,
|
||||
storage_scope_to_update.fetcher,
|
||||
storage_scope_to_update.collector,
|
||||
storage_scope_to_update.scope_key)
|
||||
|
||||
if scope_key:
|
||||
db_scope.scope_key = scope_key
|
||||
if fetcher:
|
||||
db_scope.fetcher = fetcher
|
||||
if collector:
|
||||
db_scope.collector = collector
|
||||
if active is not None and active != db_scope.active:
|
||||
db_scope.active = active
|
||||
if scope_key:
|
||||
db_scope.scope_key = scope_key
|
||||
if fetcher:
|
||||
db_scope.fetcher = fetcher
|
||||
if collector:
|
||||
db_scope.collector = collector
|
||||
if active is not None and active != db_scope.active:
|
||||
db_scope.active = active
|
||||
|
||||
now = tzutils.localized_now()
|
||||
db_scope.scope_activation_toggle_date = tzutils.local_to_utc(
|
||||
now, naive=True)
|
||||
now = tzutils.localized_now()
|
||||
db_scope.scope_activation_toggle_date = tzutils.local_to_utc(
|
||||
now, naive=True)
|
||||
|
||||
session.commit()
|
||||
session.close()
|
||||
session.commit()
|
||||
|
||||
def is_storage_scope_active(self, identifier, fetcher=None,
|
||||
collector=None, scope_key=None):
|
||||
@ -334,11 +317,9 @@ class StateManager(object):
|
||||
:type scope_key: str
|
||||
:rtype: datetime.datetime
|
||||
"""
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
r = self._get_db_item(
|
||||
session, identifier, fetcher, collector, scope_key)
|
||||
session.close()
|
||||
with db.session_for_read() as session:
|
||||
r = self._get_db_item(
|
||||
session, identifier, fetcher, collector, scope_key)
|
||||
|
||||
return r.active
|
||||
|
||||
@ -365,23 +346,21 @@ class ReprocessingSchedulerDb(object):
|
||||
projection. The ordering field will be the `id`.
|
||||
:type order: str
|
||||
"""
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
with db.session_for_read() as session:
|
||||
|
||||
query = utils.model_query(self.model, session)
|
||||
query = utils.model_query(self.model, session)
|
||||
|
||||
if identifier:
|
||||
query = query.filter(self.model.identifier.in_(identifier))
|
||||
if remove_finished:
|
||||
query = self.remove_finished_processing_schedules(query)
|
||||
if order:
|
||||
query = query.order_by(sql.text("id %s" % order))
|
||||
if identifier:
|
||||
query = query.filter(self.model.identifier.in_(identifier))
|
||||
if remove_finished:
|
||||
query = self.remove_finished_processing_schedules(query)
|
||||
if order:
|
||||
query = query.order_by(sql.text("id %s" % order))
|
||||
|
||||
query = apply_offset_and_limit(limit, offset, query)
|
||||
query = apply_offset_and_limit(limit, offset, query)
|
||||
|
||||
result_set = query.all()
|
||||
result_set = query.all()
|
||||
|
||||
session.close()
|
||||
return result_set
|
||||
|
||||
def remove_finished_processing_schedules(self, query):
|
||||
@ -398,13 +377,10 @@ class ReprocessingSchedulerDb(object):
|
||||
:type reprocessing_scheduler: models.ReprocessingScheduler
|
||||
"""
|
||||
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
with db.session_for_write() as session:
|
||||
|
||||
session.add(reprocessing_scheduler)
|
||||
session.commit()
|
||||
|
||||
session.close()
|
||||
session.add(reprocessing_scheduler)
|
||||
session.commit()
|
||||
|
||||
def get_from_db(self, identifier=None, start_reprocess_time=None,
|
||||
end_reprocess_time=None):
|
||||
@ -419,12 +395,10 @@ class ReprocessingSchedulerDb(object):
|
||||
reprocessing schedule
|
||||
:type end_reprocess_time: datetime.datetime
|
||||
"""
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
with db.session_for_read() as session:
|
||||
|
||||
result_set = self._get_db_item(
|
||||
end_reprocess_time, identifier, session, start_reprocess_time)
|
||||
session.close()
|
||||
result_set = self._get_db_item(
|
||||
end_reprocess_time, identifier, session, start_reprocess_time)
|
||||
|
||||
return result_set
|
||||
|
||||
@ -459,22 +433,21 @@ class ReprocessingSchedulerDb(object):
|
||||
:type new_current_time_stamp: datetime.datetime
|
||||
"""
|
||||
|
||||
session = db.get_session()
|
||||
session.begin()
|
||||
with db.session_for_write() as session:
|
||||
|
||||
result_set = self._get_db_item(
|
||||
end_reprocess_time, identifier, session, start_reprocess_time)
|
||||
result_set = self._get_db_item(
|
||||
end_reprocess_time, identifier, session, start_reprocess_time)
|
||||
|
||||
if not result_set:
|
||||
LOG.warning("Trying to update current time to [%s] for identifier "
|
||||
"[%s] and reprocessing range [start=%, end=%s], but "
|
||||
"we could not find a this task in the database.",
|
||||
new_current_time_stamp, identifier,
|
||||
start_reprocess_time, end_reprocess_time)
|
||||
return
|
||||
new_current_time_stamp = tzutils.local_to_utc(
|
||||
new_current_time_stamp, naive=True)
|
||||
if not result_set:
|
||||
LOG.warning("Trying to update current time to [%s] for "
|
||||
"identifier [%s] and reprocessing range [start=%, "
|
||||
"end=%s], but we could not find a this task in the"
|
||||
" database.",
|
||||
new_current_time_stamp, identifier,
|
||||
start_reprocess_time, end_reprocess_time)
|
||||
return
|
||||
new_current_time_stamp = tzutils.local_to_utc(
|
||||
new_current_time_stamp, naive=True)
|
||||
|
||||
result_set.current_reprocess_time = new_current_time_stamp
|
||||
session.commit()
|
||||
session.close()
|
||||
result_set.current_reprocess_time = new_current_time_stamp
|
||||
session.commit()
|
||||
|
@ -90,8 +90,10 @@ class TestCase(testscenarios.TestWithScenarios, base.BaseTestCase):
|
||||
self.app_context.push()
|
||||
|
||||
def tearDown(self):
|
||||
db.get_engine().dispose()
|
||||
self.auth.stop()
|
||||
self.session.stop()
|
||||
self.app_context.pop()
|
||||
super(TestCase, self).tearDown()
|
||||
with db.session_for_write() as session:
|
||||
engine = session.get_bind()
|
||||
engine.dispose()
|
||||
self.auth.stop()
|
||||
self.session.stop()
|
||||
self.app_context.pop()
|
||||
super(TestCase, self).tearDown()
|
||||
|
@ -217,7 +217,9 @@ class ConfigFixture(fixture.GabbiFixture):
|
||||
def stop_fixture(self):
|
||||
if self.conf:
|
||||
self.conf.reset()
|
||||
db.get_engine().dispose()
|
||||
with db.session_for_write() as session:
|
||||
engine = session.get_bind()
|
||||
engine.dispose()
|
||||
|
||||
|
||||
class ConfigFixtureStorageV2(ConfigFixture):
|
||||
@ -341,11 +343,11 @@ class BaseStorageDataFixture(fixture.GabbiFixture):
|
||||
|
||||
def stop_fixture(self):
|
||||
model = models.RatedDataFrame
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
model,
|
||||
session)
|
||||
q.delete()
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
model,
|
||||
session)
|
||||
q.delete()
|
||||
|
||||
|
||||
class StorageDataFixture(BaseStorageDataFixture):
|
||||
@ -405,11 +407,11 @@ class ScopeStateFixture(fixture.GabbiFixture):
|
||||
d[0], d[1], fetcher=d[2], collector=d[3], scope_key=d[4])
|
||||
|
||||
def stop_fixture(self):
|
||||
session = db.get_session()
|
||||
q = utils.model_query(
|
||||
self.sm.model,
|
||||
session)
|
||||
q.delete()
|
||||
with db.session_for_write() as session:
|
||||
q = utils.model_query(
|
||||
self.sm.model,
|
||||
session)
|
||||
q.delete()
|
||||
|
||||
|
||||
class CORSConfigFixture(fixture.GabbiFixture):
|
||||
|
@ -28,7 +28,8 @@ class StateManagerTest(tests.TestCase):
|
||||
|
||||
``filter()`` can be called any number of times, followed by first(),
|
||||
which will cycle over the ``output`` parameter passed to the
|
||||
constructor. The ``first_called`` attributes
|
||||
constructor. The ``first_called`` attribute tracks how many times
|
||||
first() is called.
|
||||
"""
|
||||
def __init__(self, output, *args, **kwargs):
|
||||
super(StateManagerTest.QueryMock, self).__init__(*args, **kwargs)
|
||||
@ -80,7 +81,7 @@ class StateManagerTest(tests.TestCase):
|
||||
self._test_x_state_does_update_columns(self._state.get_state)
|
||||
|
||||
def test_set_state_does_update_columns(self):
|
||||
with mock.patch('cloudkitty.db.get_session'):
|
||||
with mock.patch('cloudkitty.db.session_for_write'):
|
||||
self._test_x_state_does_update_columns(
|
||||
lambda x: self._state.set_state(x, datetime(2042, 1, 1)))
|
||||
|
||||
@ -101,7 +102,7 @@ class StateManagerTest(tests.TestCase):
|
||||
self._test_x_state_no_column_update(self._state.get_state)
|
||||
|
||||
def test_set_state_no_column_update(self):
|
||||
with mock.patch('cloudkitty.db.get_session'):
|
||||
with mock.patch('cloudkitty.db.session_for_write'):
|
||||
self._test_x_state_no_column_update(
|
||||
lambda x: self._state.set_state(x, datetime(2042, 1, 1)))
|
||||
|
||||
@ -111,8 +112,10 @@ class StateManagerTest(tests.TestCase):
|
||||
self._get_r_mock('a', 'b', 'c', state))
|
||||
with mock.patch(
|
||||
'oslo_db.sqlalchemy.utils.model_query',
|
||||
new=query_mock), mock.patch('cloudkitty.db.get_session') as sm:
|
||||
sm.return_value = session_mock = mock.MagicMock()
|
||||
new=query_mock), mock.patch(
|
||||
'cloudkitty.db.session_for_write') as sm:
|
||||
sm.return_value.__enter__.return_value = session_mock = \
|
||||
mock.MagicMock()
|
||||
self._state.set_state('fake_identifier', state)
|
||||
session_mock.commit.assert_not_called()
|
||||
session_mock.add.assert_not_called()
|
||||
@ -123,8 +126,10 @@ class StateManagerTest(tests.TestCase):
|
||||
new_state = datetime(2042, 1, 1)
|
||||
with mock.patch(
|
||||
'oslo_db.sqlalchemy.utils.model_query',
|
||||
new=query_mock), mock.patch('cloudkitty.db.get_session') as sm:
|
||||
sm.return_value = session_mock = mock.MagicMock()
|
||||
new=query_mock), mock.patch(
|
||||
'cloudkitty.db.session_for_write') as sm:
|
||||
sm.return_value.__enter__.return_value = session_mock = \
|
||||
mock.MagicMock()
|
||||
self.assertNotEqual(r_mock.state, new_state)
|
||||
self._state.set_state('fake_identifier', new_state)
|
||||
self.assertEqual(r_mock.last_processed_timestamp, new_state)
|
||||
|
Loading…
x
Reference in New Issue
Block a user