lsst.dax.apdb.scripts.create_cassandra Namespace Reference

Functions
None	create_cassandra (str output_config, str|None ra_dec_columns, str|None table_options=None, **Any kwargs)
CreateTableOptions|None	_read_table_options (str|None table_options=None)

Function Documentation

_read_table_options()

CreateTableOptions | None lsst.dax.apdb.scripts.create_cassandra._read_table_options ( str | None table_options = None )

protected

Read contents of the table options file.

Definition at line 60 of file create_cassandra.py.

def _read_table_options(table_options: str | None = None) -> CreateTableOptions | None:
    """Read contents of the table options file."""
    if not table_options:
        return None
 
    res = ResourcePath(table_options)
    content = res.read()
    stream = io.BytesIO(content)
    if model_data := yaml.load(stream, Loader=yaml.SafeLoader):
        return CreateTableOptions.model_validate(model_data)
    return None

create_cassandra()

None lsst.dax.apdb.scripts.create_cassandra.create_cassandra	(	str	output_config,
		str | None	ra_dec_columns,
		str | None	table_options = None,
		**Any	kwargs )

Create new APDB instance in Cassandra cluster.

Parameters
----------
output_config : `str`
    Name of the file to write APDB configuration.
ra_dec_columns : `str` or `None`
    Comma-separated list of names for ra/dec columns in DiaObject table.
table_options : `str`, optional
    Path or URI of the YAML file with table options.
**kwargs
    Keyword arguments passed to `ApdbCassandra.init_database` method.

Definition at line 35 of file create_cassandra.py.

) -> None:
    """Create new APDB instance in Cassandra cluster.
 
    Parameters
    ----------
    output_config : `str`
        Name of the file to write APDB configuration.
    ra_dec_columns : `str` or `None`
        Comma-separated list of names for ra/dec columns in DiaObject table.
    table_options : `str`, optional
        Path or URI of the YAML file with table options.
    **kwargs
        Keyword arguments passed to `ApdbCassandra.init_database` method.
    """
    options = _read_table_options(table_options)
    ra_dec_list: list[str] | None = None
    if ra_dec_columns:
        ra_dec_list = ra_dec_columns.split(",")
    kwargs["hosts"] = kwargs.pop("host")
    config = ApdbCassandra.init_database(ra_dec_columns=ra_dec_list, table_options=options, **kwargs)
    config.save(output_config)