uncloud/schemas.py
2020-01-20 12:30:12 +05:00

134 lines
4.1 KiB
Python

import logging
import config
from helper import resolve_product_usable_id
etcd_client = config.etcd_client
class ValidationException(Exception):
"""Validation Error"""
class Field:
def __init__(self, _name, _type, _value=None, validators=None):
if validators is None:
validators = []
assert isinstance(validators, list)
self.name = _name
self.value = _value
self.type = _type
self.validators = validators
def is_valid(self):
if not isinstance(self.value, self.type):
try:
self.value = self.type(self.value)
except Exception:
raise ValidationException("Incorrect Type for '{}' field".format(self.name))
for validator in self.validators:
validator()
def __repr__(self):
return self.name
class BaseSchema:
def __init__(self):
self.fields = [getattr(self, field) for field in dir(self) if isinstance(getattr(self, field), Field)]
def validation(self):
# custom validation is optional
return True
def is_valid(self):
for field in self.fields:
field.is_valid()
for parent in self.__class__.__bases__:
parent.validation(self)
self.validation()
for field in self.fields:
setattr(self, field.name, field.value)
def return_data(self):
return {
field.name: field.value
for field in self.fields
}
def get(dictionary: dict, key: str, return_default=False, default=None):
if dictionary is None:
raise ValidationException('No data provided at all.')
try:
value = dictionary[key]
except KeyError:
if return_default:
return default
raise ValidationException("Missing data for '{}' field.".format(key))
else:
return value
class AddProductSchema(BaseSchema):
def __init__(self, data):
self.email = Field('email', str, get(data, 'email'))
self.password = Field('password', str, get(data, 'password'))
self.specs = Field('specs', dict, get(data, 'specs'))
super().__init__()
class UserRegisterPaymentSchema(BaseSchema):
def __init__(self, data):
self.email = Field('email', str, get(data, 'email'))
self.password = Field('password', str, get(data, 'password'))
self.card_number = Field('card_number', str, get(data, 'card_number'))
self.cvc = Field('cvc', str, get(data, 'cvc'))
self.expiry_year = Field('expiry_year', int, get(data, 'expiry_year'))
self.expiry_month = Field('expiry_month', int, get(data, 'expiry_month'))
self.card_holder_name = Field('card_holder_name', str, get(data, 'card_holder_name'))
super().__init__()
class ProductOrderSchema(BaseSchema):
def __init__(self, data):
self.email = Field('email', str, get(data, 'email'))
self.password = Field('password', str, get(data, 'password'))
self.product_id = Field('product_id', str, get(data, 'product_id'), validators=[self.product_id_validation])
super().__init__()
def product_id_validation(self):
product_uuid = resolve_product_usable_id(self.product_id.value, etcd_client)
if product_uuid:
self.product_id.value = product_uuid
else:
raise ValidationException('Invalid Product ID')
class OrderListSchema(BaseSchema):
def __init__(self, data):
self.email = Field('email', str, get(data, 'email'))
self.password = Field('password', str, get(data, 'password'))
super().__init__()
def make_return_message(err, status_code=200):
logging.debug('message: {}'.format(str(err)))
return {'message': str(err)}, status_code
def create_schema(specification, data):
fields = {}
for feature_name, feature_detail in specification['features'].items():
if not feature_detail['constant']:
fields[feature_name] = Field(feature_name, eval(feature_detail['unit']['type']), get(data, feature_name))
return type('{}Schema'.format(specification['name']), (BaseSchema,), fields)