# -*- coding: utf-8 -*-
import logging
import warnings
import six
from brewtils.models import (
System, Instance, Command, Parameter, Request, PatchOperation, Choices,
LoggingConfig, Event, Queue, Principal, Role, RefreshToken, Job,
RequestTemplate, DateTrigger, CronTrigger, IntervalTrigger
)
from brewtils.schemas import (
SystemSchema, InstanceSchema, CommandSchema, ParameterSchema, RequestSchema,
PatchSchema, LoggingConfigSchema, EventSchema, QueueSchema, PrincipalSchema,
RoleSchema, RefreshTokenSchema, JobSchema
)
[docs]class SchemaParser(object):
"""Serialize and deserialize Brewtils models"""
_models = {
'SystemSchema': System,
'InstanceSchema': Instance,
'CommandSchema': Command,
'ParameterSchema': Parameter,
'RequestTemplateSchema': RequestTemplate,
'RequestSchema': Request,
'PatchSchema': PatchOperation,
'ChoicesSchema': Choices,
'LoggingConfigSchema': LoggingConfig,
'EventSchema': Event,
'QueueSchema': Queue,
'PrincipalSchema': Principal,
'RoleSchema': Role,
'RefreshTokenSchema': RefreshToken,
'JobSchema': Job,
'DateTriggerSchema': DateTrigger,
'IntervalTriggerSchema': IntervalTrigger,
'CronTriggerSchema': CronTrigger,
}
logger = logging.getLogger(__name__)
# Deserialization methods
[docs] @classmethod
def parse_system(cls, system, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a system model object
:param system: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A System object
"""
return cls._do_parse(system, SystemSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_instance(cls, instance, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to an instance model object
:param instance: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: An Instance object
"""
return cls._do_parse(instance, InstanceSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_command(cls, command, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a command model object
:param command: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A Command object
"""
return cls._do_parse(command, CommandSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_parameter(cls, parameter, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a parameter model object
:param parameter: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: An Parameter object
"""
return cls._do_parse(parameter, ParameterSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_request(cls, request, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a request model object
:param request: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A Request object
"""
return cls._do_parse(request, RequestSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_patch(cls, patch, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a patch model object
Note: for our patches, many is _always_ set to True. We will always return a list
from this method.
:param patch: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A PatchOperation object
"""
if not kwargs.pop('many', True):
cls.logger.warning("A patch object should always be wrapped as a list of objects. "
"Thus, parsing will always return a list. You specified many as "
"False, this is being ignored and a list "
"will be returned anyway.")
return cls._do_parse(patch, PatchSchema(many=True, **kwargs), from_string=from_string)
[docs] @classmethod
def parse_logging_config(cls, logging_config, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a logging config model object
Note: for our logging_config, many is _always_ set to False. We will always return a dict
from this method.
:param logging_config: The raw input
:param from_string: True if 'input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A LoggingConfig object
"""
if kwargs.pop('many', False):
cls.logger.warning("A logging config object should never be wrapped as a list of "
"objects. Thus, parsing will always return a dict. You specified "
"many as True, this is being ignored and a dict will be returned "
"anyway.")
return cls._do_parse(logging_config,
LoggingConfigSchema(many=False, **kwargs),
from_string=from_string)
[docs] @classmethod
def parse_event(cls, event, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to an event model object
:param event: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: An Event object
"""
return cls._do_parse(event, EventSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_queue(cls, queue, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a queue model object
:param queue: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A Queue object
"""
return cls._do_parse(queue, QueueSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_principal(cls, principal, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a principal model object
:param principal: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A Principal object
"""
return cls._do_parse(principal, PrincipalSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_role(cls, role, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a role model object
:param role: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A Role object
"""
return cls._do_parse(role, RoleSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_refresh_token(cls, refresh_token, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a refresh token object
:param refresh_token: The raw input
:param from_string: True if input is a JSON string, False if a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: A RefreshToken object
"""
return cls._do_parse(refresh_token, RefreshTokenSchema(**kwargs), from_string=from_string)
[docs] @classmethod
def parse_job(cls, job, from_string=False, **kwargs):
"""Convert raw JSON string or dictionary to a job model object
Args:
job: Raw input
from_string: True if input is a JSON string, False if a dictionary
**kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
Returns:
A Job object.
"""
return cls._do_parse(job, JobSchema(**kwargs), from_string=from_string)
@classmethod
def _do_parse(cls, data, schema, from_string=False):
if from_string and not isinstance(data, six.string_types):
raise TypeError("When from_string=True data must be a string-type")
schema.context['models'] = cls._models
return schema.loads(data).data if from_string else schema.load(data).data
# Serialization methods
[docs] @classmethod
def serialize_system(cls, system, to_string=True, include_commands=True, **kwargs):
"""Convert a system model into serialized form
:param system: The system object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param include_commands: True if the system's command list should be included
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of system
"""
if not include_commands:
if 'exclude' in kwargs:
kwargs['exclude'] += ('commands', )
else:
kwargs['exclude'] = ('commands', )
return cls._do_serialize(SystemSchema(**kwargs), system, to_string)
[docs] @classmethod
def serialize_instance(cls, instance, to_string=True, **kwargs):
"""Convert an instance model into serialized form
:param instance: The instance object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of instance
"""
return cls._do_serialize(InstanceSchema(**kwargs), instance, to_string)
[docs] @classmethod
def serialize_command(cls, command, to_string=True, **kwargs):
"""Convert a command model into serialized form
:param command: The command object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of command
"""
return cls._do_serialize(CommandSchema(**kwargs), command, to_string)
[docs] @classmethod
def serialize_parameter(cls, parameter, to_string=True, **kwargs):
"""Convert a parameter model into serialized form
:param parameter: The parameter object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of parameter
"""
return cls._do_serialize(ParameterSchema(**kwargs), parameter, to_string)
[docs] @classmethod
def serialize_request(cls, request, to_string=True, **kwargs):
"""Convert a request model into serialized form
:param request: The request object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of request
"""
return cls._do_serialize(RequestSchema(**kwargs), request, to_string)
[docs] @classmethod
def serialize_patch(cls, patch, to_string=True, **kwargs):
"""Convert a patch model into serialized form
:param patch: The patch object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of patch
"""
return cls._do_serialize(PatchSchema(**kwargs), patch, to_string)
[docs] @classmethod
def serialize_logging_config(cls, logging_config, to_string=True, **kwargs):
"""Convert a logging config model into serialize form
:param logging_config: The logging config object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of logging config
"""
return cls._do_serialize(LoggingConfigSchema(**kwargs), logging_config, to_string)
[docs] @classmethod
def serialize_event(cls, event, to_string=True, **kwargs):
"""Convert a logging config model into serialized form
:param event: The event object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of event
"""
return cls._do_serialize(EventSchema(**kwargs), event, to_string)
[docs] @classmethod
def serialize_queue(cls, queue, to_string=True, **kwargs):
"""Convert a queue model into serialized form
:param queue: The queue object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation of queue
"""
return cls._do_serialize(QueueSchema(**kwargs), queue, to_string)
[docs] @classmethod
def serialize_principal(cls, principal, to_string=True, **kwargs):
"""Convert a principal model into serialized form
:param principal: The principal object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation
"""
return cls._do_serialize(PrincipalSchema(**kwargs), principal, to_string)
[docs] @classmethod
def serialize_role(cls, role, to_string=True, **kwargs):
"""Convert a role model into serialized form
:param role: The role object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation
"""
return cls._do_serialize(RoleSchema(**kwargs), role, to_string)
[docs] @classmethod
def serialize_refresh_token(cls, refresh_token, to_string=True, **kwargs):
"""Convert a role model into serialized form
:param refresh_token: The token object(s) to be serialized
:param to_string: True to generate a JSON-formatted string, False to generate a dictionary
:param kwargs: Additional parameters to be passed to the Schema (e.g. many=True)
:return: Serialized representation
"""
return cls._do_serialize(RefreshTokenSchema(**kwargs), refresh_token, to_string)
[docs] @classmethod
def serialize_job(cls, job, to_string=True, **kwargs):
"""Convert a job model into serialized form.
Args:
job: The job object(s) to be serialized.
to_string: True to generate a JSON-formatted string, False to generate a dictionary.
**kwargs: Additional parameters to be passed to the shcema (e.g. many=True)
Returns:
Serialize representation of job.
"""
return cls._do_serialize(JobSchema(**kwargs), job, to_string)
@staticmethod
def _do_serialize(schema, data, to_string):
return schema.dumps(data).data if to_string else schema.dump(data).data
[docs]class BrewmasterSchemaParser(SchemaParser):
def __init__(self):
warnings.warn("Reference made to 'BrewmasterSchemaParser'. This name will be removed in "
"version 3.0, please use 'SchemaParser' instead.",
DeprecationWarning, stacklevel=2)
super(BrewmasterSchemaParser, self).__init__()