Upload Data Products to OpenMetadata

Question

I need to build a python application to interact with the OpenMetadata API (specifically, to upload a data in YAML format). However, I’ve encountered some intrinsic connections between the OpenMetadata entities (Tables, Columns, DatabaseSchema, ...) that I’m unable to decouple in order to create modular code (i.e, code that adheres to the SOLID principles)

As you can see, the TableCreator depends on other creators (ex: ColumnCreator, DatabaseSchemaCreator) and also requires the metadata client to perform searches for entities which are necessary to build the request.

What do you recommend to make it more SOLID compliant?

class GenericEntityCreator(ABC): def __init__(self, entity_handler: Optional['EntityHandler']=None): self._entity_handler = entity_handler @abstractmethod def create_entity(self, data: Any) -> T: pass 

class EntityHandler: def __init__(self, metadata_client: OpenMetadataClient): self.metadata_client = metadata_client self._creators: Dict[Type, GenericEntityCreator] = {} self._initialize_creators() def _initialize_creators(self) -> None: self._creators.update({ Table: TableCreator(self), Column: ColumnCreator(self), DatabaseSchema: DatabaseSchemaCreator(self), }) def get_creator(self, entity_type: Type[T]) -> GenericEntityCreator: creator = self._creators.get(entity_type) if not creator: raise ValueError(f"No creator registered for entity type {entity_type}") return creator 

class TableCreator(GenericEntityCreator): def get_search_string(self, data: TableData) -> str: return (f"{data.origin.capitalize()}." f"{data.platforms[data.origin].database}." f"{data.platforms[data.origin].databaseSchema}." f"{data.name.root}") def create_entity(self, data: TableData) -> CreateTableRequest: column_creator = self._entity_handler.get_creator(Column) columns = [column_creator.create_entity(col) for col in data.columns] db_schema_creator = self._entity_handler.get_creator(DatabaseSchema) search_string = db_schema_creator.get_search_string(data) database_schema = self._entity_handler.metadata_client.get_entity_from_es(entity_type=DatabaseSchema, fqn_search_string=search_string) return CreateTableRequest( name=EntityName(root=data.name.root), description=data.description, tableType=TableType("Regular"), databaseSchema=FullyQualifiedEntityName(root=database_schema.fullyQualifiedName.root), columns=columns ) 

--main.py entity_handler = EntityHandler(omd_client) table_creator = entity_handler.get_creator(Table) table_creator.create_entity(data)