(self, source: Union[str, Path, TextIO] = None)
return schema
Load schemas from a local file

        These files will be treated as local schemas, and will not be sent to the bus.
        This can be useful for validation during development and testing.
Load schemas from a local file
Load schemas from a local file
def load_local(self, source: Union[str, Path, TextIO] = None):
        """Load schemas from a local file

        These files will be treated as local schemas, and will not be sent to the bus.
        This can be useful for validation during development and testing.
        """
        if isinstance(source, str):
            source = Path(source)

        def _load_schema(path, file_data):
            try:
                return json.loads(file_data)
            except JSONDecodeError as e:
                raise InvalidSchema("Could not parse schema file {}: {}".format(path, e.msg))

        if source is None:
            # No source, read from stdin
            schema = _load_schema("[stdin]", sys.stdin.read())
        elif hasattr(source, "is_dir") and source.is_dir():
            # Read each json file in directory
            schemas = []
            for file_path in source.glob("*.json"):
                schemas.append(_load_schema(file_path, file_path.read_text(encoding="utf8")))
            schema = ChainMap(*schemas)
        elif hasattr(source, "read"):
            # Read file handle
            schema = _load_schema(source.name, source.read())
        elif hasattr(source, "read_text"):
            # Read pathlib Path
            schema = _load_schema(source.name, source.read_text())
        else:
            raise InvalidSchema(
                "Did not recognise provided source as either a "
                "directory path, file path, or file handle: {}".format(source)
            )

        for api_name, api_schema in schema.items():
            self.local_schemas[api_name] = api_schema

        return schema
def load_local ( self , source : Union [ str , Path , TextIO ] = None ) : if isinstance ( source , str ) : source = Path ( source ) def _load_schema ( path , file_data ) : try : return json . loads ( file_data ) except JSONDecodeError as e : raise InvalidSchema ( "Could not parse schema file {}: {}" . format ( path , e . msg ) ) if source is None : # No source, read from stdin schema = _load_schema ( "[stdin]" , sys . stdin . read ( ) ) elif hasattr ( source , "is_dir" ) and source . is_dir ( ) : # Read each json file in directory schemas = [ ] for file_path in source . glob ( "*.json" ) : schemas . append ( _load_schema ( file_path , file_path . read_text ( encoding = "utf8" ) ) ) schema = ChainMap ( * schemas ) elif hasattr ( source , "read" ) : # Read file handle schema = _load_schema ( source . name , source . read ( ) ) elif hasattr ( source , "read_text" ) : # Read pathlib Path schema = _load_schema ( source . name , source . read_text ( ) ) else : raise InvalidSchema ( "Did not recognise provided source as either a " "directory path, file path, or file handle: {}" . format ( source ) ) for api_name , api_schema in schema . items ( ) : self . local_schemas [ api_name ] = api_schema return schema