I have defined a new custom block to connect to MongoDB as follows:
class MongoDBConnection(Block):
host: str
username: str
password: SecretStr
authSource: Optional[str] = 'admin'
def get_client(self) -> MongoClient:
return MongoClient(
host=self.host,
username=self.username,
password=self.password.get_secret_value(),
authSource=self.authSource
)
I have registered the block using the cli as follows:
prefect block register --file mongodb_connection.py
I have no issues using this block if I import the class from mongodb_connection.py
and call MongoDBConnection.load(<name>)
. However, I would like to use the generic Block.load(<slug>/<name>)
pattern to avoid messy imports when sharing custom block definitions between flows.
I get the following error when I try to use the generic load pattern:
>>> mongo_block = Block.load('mongodbconnection/<name>')
KeyError: "No class found for dispatch key 'mongodbconnection' in registry for type 'Block'."
How would I go about using my custom blocks without needing messy imports?