How to use the dagster.check function in dagster

def create_serialization_step(solid, output_def, prev_subplan):
    check.inst_param(solid, 'solid', Solid)
    check.inst_param(output_def, 'output_def', OutputDefinition)
    check.inst_param(prev_subplan, 'prev_subplan', ExecutionValueSubPlan)

    return ExecutionStep(
        key='serialize.' + solid.name + '.' + output_def.name,
        step_inputs=[
            StepInput(
                name=SERIALIZE_INPUT,
                dagster_type=output_def.dagster_type,
                prev_output_handle=prev_subplan.terminal_step_output_handle,
            )
        ],
        step_outputs=[StepOutput(name=SERIALIZE_OUTPUT, dagster_type=output_def.dagster_type)],
        compute_fn=_create_serialization_lambda(solid, output_def),
        tag=StepTag.SERIALIZE,
        solid=solid,
    )

def _get_pipelines(graphene_info):
    check.inst_param(graphene_info, 'graphene_info', ResolveInfo)

    def process_pipelines(repository):
        try:
            pipeline_instances = []
            for pipeline_def in repository.get_all_pipelines():
                pipeline_instances.append(graphene_info.schema.type_named('Pipeline')(pipeline_def))
            return graphene_info.schema.type_named('PipelineConnection')(
                nodes=sorted(pipeline_instances, key=lambda pipeline: pipeline.name)
            )
        except DagsterInvalidDefinitionError:
            return EitherError(
                graphene_info.schema.type_named('InvalidDefinitionError')(
                    serializable_error_info_from_exc_info(sys.exc_info())
                )
            )

def delete_schedule(self, schedule):
        check.inst_param(schedule, 'schedule', Schedule)
        self._schedules.pop(schedule.name)
        self._delete_schedule_file(schedule)

def check_supports_sql_alchemy_resource(context):
    check.inst_param(context, 'context', ExecutionContext)
    check.invariant(context.resources is not None)
    check.invariant(
        hasattr(context.resources, 'sa'),
        'Resources must have sa property be an object of SqlAlchemyResource',
    )
    check.inst(
        context.resources.sa,
        SqlAlchemyResource,
        'Resources must have sa property be an object of SqlAlchemyResource',
    )
    return context

def __new__(cls, field_names, field_defs):
        return super(MissingFieldsErrorData, cls).__new__(
            cls,
            check.list_param(field_names, 'field_names', of_type=str),
            [check_field_param(field_def, 'field_defs') for field_def in field_defs],
        )

def user_code_context_manager(user_fn, error_cls, msg):
    '''Wraps the output of a user provided function that may yield or return a value and
    returns a generator that asserts it only yields a single value.
    '''
    check.callable_param(user_fn, 'user_fn')
    check.subclass_param(error_cls, 'error_cls', DagsterUserCodeExecutionError)

    with user_code_error_boundary(error_cls, msg):
        thing_or_gen = user_fn()
        gen = _ensure_gen(thing_or_gen)

        try:
            thing = next(gen)
        except StopIteration:
            check.failed('Must yield one item. You did not yield anything.')

        yield thing

        stopped = False

        try:
            next(gen)

def from_event_record(graphene_info, event_record, dauphin_pipeline, execution_plan):
    check.inst_param(event_record, 'event_record', EventRecord)
    check.inst_param(
        dauphin_pipeline, 'dauphin_pipeline', graphene_info.schema.type_named('Pipeline')
    )
    check.inst_param(execution_plan, 'execution_plan', ExecutionPlan)

    if event_record.is_dagster_event:
        return from_dagster_event_record(
            graphene_info, event_record, dauphin_pipeline, execution_plan
        )
    else:
        return graphene_info.schema.type_named('LogMessageEvent')(
            **construct_basic_params(graphene_info, event_record, execution_plan)
        )

user code error boundary around invoked user-space code. These terminate
        the computation immediately (by re-raising) even if raise_on_error is false.

    If the raise_on_error option is set to True, these errors are reraised and surfaced
    to the user. This is mostly to get sensible errors in test and ad-hoc contexts, rather
    than forcing the user to wade through the PipelineExecutionResult API in order to find
    the step that errored.

    For tools, however, this option should be false, and a sensible error message
    signaled to the user within that tool.
    '''

    check.inst_param(step_context, 'step_context', SystemStepExecutionContext)

    try:
        for step_event in check.generator(_core_dagster_event_sequence_for_step(step_context)):
            yield step_event

    # case (1) in top comment
    except DagsterUserCodeExecutionError as dagster_user_error:  # case (1) above
        yield _step_failure_event_from_exc_info(
            step_context,
            dagster_user_error.original_exc_info,
            UserFailureData(
                label='intentional-failure',
                description=dagster_user_error.user_specified_failure.description,
                metadata_entries=dagster_user_error.user_specified_failure.metadata_entries,
            )
            if dagster_user_error.is_user_specified_failure
            else None,
        )

def create_single_solid_env_from_arg_dicts(solid, arg_dicts):
    check.inst_param(solid, 'solid', SolidDefinition)
    check.dict_param(arg_dicts, 'arg_dicts', key_type=str, value_type=dict)

    input_to_source_type = {}
    for input_def in solid.inputs:
        check.invariant(len(input_def.sources) == 1)
        input_to_source_type[input_def.name] = input_def.sources[0].source_type

    sources = {}

    for input_name, arg_dict in arg_dicts.items():
        source_name = input_to_source_type[input_name]
        sources[input_name] = config.Source(name=source_name, args=arg_dict)

    return config.Environment(sources=sources)

def __new__(cls, is_builtin=False, is_system_config=False):
        return super(ConfigTypeAttributes, cls).__new__(
            cls,
            is_builtin=check.bool_param(is_builtin, 'is_builtin'),
            is_system_config=check.bool_param(is_system_config, 'is_system_config'),
        )