Skip to content

Instantly share code, notes, and snippets.

@tomplex
Created April 26, 2020 06:28
Show Gist options
  • Save tomplex/c0df7e211908da46836a35af6ab5370f to your computer and use it in GitHub Desktop.
Save tomplex/c0df7e211908da46836a35af6ab5370f to your computer and use it in GitHub Desktop.
Use Pydantic to dynamically create specific implementations of an ABC based on input data
import abc
import typing
import pydantic
class File(pydantic.BaseModel, abc.ABC):
@classmethod
def __get_validators__(cls):
# one or more validators may be yielded which will be called in the
# order to validate the input, each validator will receive as an input
# the value returned from the previous validator
yield cls.validate
@classmethod
def validate(cls, v):
subclass_dict: typing.Dict[str, pydantic.BaseModel] = {
sbc.__name__: sbc for sbc in cls.__subclasses__()
}
match = list(set(subclass_dict.keys()).intersection(v.keys()))
if not match:
raise ValueError("Couldn't find a File type matching input: {}".format(v))
classname = match[0]
return subclass_dict[classname].parse_obj(v[classname])
@abc.abstractmethod
def open(self, *args, **kwargs) -> typing.IO:
raise NotImplementedError
class S3File(File):
bucket: str
key: str
def open(self, *args, **kwargs) -> typing.IO:
raise NotImplementedError
class LocalFile(File):
path: str
def open(self, mode="r", *args, **kwargs) -> typing.IO:
return open(self.path, mode, *args, **kwargs)
class Configuration(pydantic.BaseModel):
"""
This model will be created from the
data above using the BaseModel.parse_obj()
classmethod.
"""
source: File
data = {"source": {"LocalFile": {"path": "/tmp/test.txt"}}}
config: Configuration = Configuration.parse_obj(data)
with config.source.open() as f:
print(f.read())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment