async def METHOD_NAME(
    self,
    session: sa.orm.session,
    db: PrefectDBInterface,
    notification,
):
    try:
        orm_block_document = await session.get(
            db.BlockDocument, notification.block_document_id
        )
        if orm_block_document is None:
            self.logger.error(
                f"Missing block document {notification.block_document_id} "
                f"from policy {notification.flow_run_notification_policy_id}"
            )
            return
        from prefect.blocks.core import Block
        block = Block._from_block_document(
            await schemas.core.BlockDocument.from_orm_model(
                session=session,
                orm_block_document=orm_block_document,
                include_secrets=True,
            )
        )
        message = self.construct_notification_message(notification=notification)
        await block.notify(
            subject="Prefect flow run notification",
            body=message,
        )
        self.logger.debug(
            "Successfully sent notification for flow run"
            f" {notification.flow_run_id} from policy"
            f" {notification.flow_run_notification_policy_id}"
        )
    except Exception:
        self.logger.error(
            (
                "Error sending notification for policy"
                f" {notification.flow_run_notification_policy_id} on flow run"
                f" {notification.flow_run_id}"
            ),
            exc_info=True,
        )