How to use the hologram.FieldEncoder function in hologram

To help you get started, we’ve selected a few hologram examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github fishtown-analytics / dbt / plugins / redshift / dbt / adapters / redshift / connections.py View on Github external
import boto3

from hologram import FieldEncoder, JsonSchemaMixin
from hologram.helpers import StrEnum

from dataclasses import dataclass, field
from typing import Optional

drop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()


IAMDuration = NewType('IAMDuration', int)


class IAMDurationEncoder(FieldEncoder):
    @property
    def json_schema(self):
        return {'type': 'integer', 'minimum': 0, 'maximum': 65535}


JsonSchemaMixin.register_field_encoders({IAMDuration: IAMDurationEncoder()})


class RedshiftConnectionMethod(StrEnum):
    DATABASE = 'database'
    IAM = 'iam'


@dataclass
class RedshiftCredentials(PostgresCredentials):
    method: RedshiftConnectionMethod = RedshiftConnectionMethod.DATABASE
github fishtown-analytics / dbt / core / dbt / helper_types.py View on Github external
# never name this package "types", or mypy will crash in ugly ways
from datetime import timedelta
from numbers import Real
from typing import NewType, Dict

from hologram import (
    FieldEncoder, JsonSchemaMixin, JsonDict, ValidationError
)


Port = NewType('Port', int)


class PortEncoder(FieldEncoder):
    @property
    def json_schema(self):
        return {'type': 'integer', 'minimum': 0, 'maximum': 65535}


class TimeDeltaFieldEncoder(FieldEncoder[timedelta]):
    """Encodes timedeltas to dictionaries"""

    def to_wire(self, value: timedelta) -> float:
        return value.total_seconds()

    def to_python(self, value) -> timedelta:
        if isinstance(value, timedelta):
            return value
        try:
            return timedelta(seconds=value)
github fishtown-analytics / dbt / core / dbt / helper_types.py View on Github external
from hologram import (
    FieldEncoder, JsonSchemaMixin, JsonDict, ValidationError
)


Port = NewType('Port', int)


class PortEncoder(FieldEncoder):
    @property
    def json_schema(self):
        return {'type': 'integer', 'minimum': 0, 'maximum': 65535}


class TimeDeltaFieldEncoder(FieldEncoder[timedelta]):
    """Encodes timedeltas to dictionaries"""

    def to_wire(self, value: timedelta) -> float:
        return value.total_seconds()

    def to_python(self, value) -> timedelta:
        if isinstance(value, timedelta):
            return value
        try:
            return timedelta(seconds=value)
        except TypeError:
            raise ValidationError(
                'cannot encode {} into timedelta'.format(value)
            ) from None

    @property
github fishtown-analytics / dbt / core / dbt / helper_types.py View on Github external
def to_python(self, value) -> timedelta:
        if isinstance(value, timedelta):
            return value
        try:
            return timedelta(seconds=value)
        except TypeError:
            raise ValidationError(
                'cannot encode {} into timedelta'.format(value)
            ) from None

    @property
    def json_schema(self) -> JsonDict:
        return {'type': 'number'}


class RealEncoder(FieldEncoder):
    @property
    def json_schema(self):
        return {'type': 'number'}


class NoValue:
    """Sometimes, you want a way to say none that isn't None"""
    def __eq__(self, other):
        return isinstance(other, NoValue)


class NoValueEncoder(FieldEncoder):
    # the FieldEncoder class specifies a narrow range that only includes value
    # types (str, float, None) but we want to support something extra
    def to_wire(self, value: NoValue) -> Dict[str, str]:  # type: ignore
        return {'novalue': 'novalue'}
github fishtown-analytics / dbt / core / dbt / helper_types.py View on Github external
return {'type': 'number'}


class RealEncoder(FieldEncoder):
    @property
    def json_schema(self):
        return {'type': 'number'}


class NoValue:
    """Sometimes, you want a way to say none that isn't None"""
    def __eq__(self, other):
        return isinstance(other, NoValue)


class NoValueEncoder(FieldEncoder):
    # the FieldEncoder class specifies a narrow range that only includes value
    # types (str, float, None) but we want to support something extra
    def to_wire(self, value: NoValue) -> Dict[str, str]:  # type: ignore
        return {'novalue': 'novalue'}

    def to_python(self, value) -> NoValue:
        if (
            not isinstance(value, dict) or
            'novalue' not in value or
            value['novalue'] != 'novalue'
        ):
            raise ValidationError('Got invalid NoValue: {}'.format(value))
        return NoValue()

    @property
    def json_schema(self):