lsst.dax.apdb.tests._apdb Namespace Reference

Classes
class	ApdbSchemaUpdateTest
class	ApdbTest

Functions
Region	_make_region (tuple[float, float, float] xyz=(1.0, 1.0, -1.0))
Iterator[str]	update_schema_yaml (str schema_file, bool drop_metadata=False, str|None version=None)

Function Documentation

_make_region()

Region lsst.dax.apdb.tests._apdb._make_region ( tuple[float, float, float] xyz = (1.0, 1.0, -1.0) )

protected

Make a region to use in tests

Definition at line 66 of file _apdb.py.

def _make_region(xyz: tuple[float, float, float] = (1.0, 1.0, -1.0)) -> Region:
    """Make a region to use in tests"""
    pointing_v = UnitVector3d(*xyz)
    fov = 0.0013  # radians
    region = Circle(pointing_v, Angle(fov / 2))
    return region
 
 
@contextlib.contextmanager

update_schema_yaml()

Iterator[str] lsst.dax.apdb.tests._apdb.update_schema_yaml	(	str	schema_file,
		bool	drop_metadata = False,
		str | None	version = None )

Update schema definition and return name of the new schema file.

Parameters
----------
schema_file : `str`
    Path for the existing YAML file with APDB schema.
drop_metadata : `bool`
    If `True` then remove metadata table from the list of tables.
version : `str` or `None`
    If non-empty string then set schema version to this string, if empty
    string then remove schema version from config, if `None` - don't change
    the version in config.

Yields
------
Path for the updated configuration file.

Definition at line 75 of file _apdb.py.

) -> Iterator[str]:
    """Update schema definition and return name of the new schema file.
 
    Parameters
    ----------
    schema_file : `str`
        Path for the existing YAML file with APDB schema.
    drop_metadata : `bool`
        If `True` then remove metadata table from the list of tables.
    version : `str` or `None`
        If non-empty string then set schema version to this string, if empty
        string then remove schema version from config, if `None` - don't change
        the version in config.
 
    Yields
    ------
    Path for the updated configuration file.
    """
    with open(schema_file) as yaml_stream:
        schemas_list = list(yaml.load_all(yaml_stream, Loader=yaml.SafeLoader))
    # Edit YAML contents.
    for schema in schemas_list:
        # Optionally drop metadata table.
        if drop_metadata:
            schema["tables"] = [table for table in schema["tables"] if table["name"] != "metadata"]
        if version is not None:
            if version == "":
                del schema["version"]
            else:
                schema["version"] = version
 
    with TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
        output_path = os.path.join(tmpdir, "schema.yaml")
        with open(output_path, "w") as yaml_stream:
            yaml.dump_all(schemas_list, stream=yaml_stream)
        yield output_path