lsst.dax.apdb.scripts.partition Namespace Reference

Functions
int	partition_show_temporal (str apdb_config)
int	partition_extend_temporal (str apdb_config, str time, bool past, int max_days)
int	partition_delete_temporal (str apdb_config, str time, bool after, bool force)
bool	_confirm_delete (*, list[int] partitions, list[str] tables, Partitioner partitioner)

Function Documentation

_confirm_delete()

bool lsst.dax.apdb.scripts.partition._confirm_delete ( * , list[int] partitions, list[str] tables, Partitioner partitioner )

protected

Definition at line 175 of file partition.py.

def _confirm_delete(*, partitions: list[int], tables: list[str], partitioner: Partitioner) -> bool:
    print("Partitions to be deleted:")
    for part in partitions:
        start_time, end_time = partitioner.partition_period(part)
        print(f"   {part}: [{start_time.tai.isot}, {end_time.tai.isot})")
    answer = input("Confirm deletion y/n: ")
    return answer.strip().lower() == "y"

partition_delete_temporal()

int lsst.dax.apdb.scripts.partition.partition_delete_temporal	(	str	apdb_config,
		str	time,
		bool	after,
		bool	force )

Delete some temporal partitions.

Parameters
----------
apdb_config : `str`
    URL for APDB configuration file.
time : `str`
    Timestamps in ISOT format and TAI scale. Partition that includes this
    time is not deleted.
after : `bool`
    If `True` then delete partitions after the specified time. Default is
    to delete partitions before this time.
force : `bool`
    If `True` then do not ask confirmation.

Definition at line 119 of file partition.py.

def partition_delete_temporal(apdb_config: str, time: str, after: bool, force: bool) -> int:
    """Delete some temporal partitions.
 
    Parameters
    ----------
    apdb_config : `str`
        URL for APDB configuration file.
    time : `str`
        Timestamps in ISOT format and TAI scale. Partition that includes this
        time is not deleted.
    after : `bool`
        If `True` then delete partitions after the specified time. Default is
        to delete partitions before this time.
    force : `bool`
        If `True` then do not ask confirmation.
    """
    try:
        astro_time = astropy.time.Time(time, format="isot", scale="tai")
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
 
    apdb = Apdb.from_uri(apdb_config)
    if not isinstance(apdb, ApdbCassandra):
        print("ERROR: Non-Cassandra APDB does not use time-partitioned tables.", file=sys.stderr)
        return 1
 
    admin = apdb.admin
    try:
        result = admin.delete_time_partitions(
            astro_time, after=after, confirm=None if force else _confirm_delete
        )
    except (TypeError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
 
    part_range = admin.time_partitions()
    start_time, _ = admin.partitioner.partition_period(part_range.start)
    _, end_time = admin.partitioner.partition_period(part_range.end)
 
    if result:
        print(
            "Time partitions succesfully deleted.\n"
            f"New time partition range: {part_range.start} - {part_range.end} "
            f"[{start_time.tai.isot}, {end_time.tai.isot})"
        )
    else:
        print(
            "Time partitions were not deleted.\n"
            f"Current time partition range: {part_range.start} - {part_range.end} "
            f"[{start_time.tai.isot}, {end_time.tai.isot})"
        )
 
    return 0
 
 

partition_extend_temporal()

int lsst.dax.apdb.scripts.partition.partition_extend_temporal	(	str	apdb_config,
		str	time,
		bool	past,
		int	max_days )

Extend the range of temporal partitions.

Parameters
----------
apdb_config : `str`
    URL for APDB configuration file.
time : `str`
    Timestamps in ISOT format and TAI scale.
past : `bool`
    If `True` extend the range in the past.
max_days : `int`
    Max. number of days for extension.

Definition at line 66 of file partition.py.

def partition_extend_temporal(apdb_config: str, time: str, past: bool, max_days: int) -> int:
    """Extend the range of temporal partitions.
 
    Parameters
    ----------
    apdb_config : `str`
        URL for APDB configuration file.
    time : `str`
        Timestamps in ISOT format and TAI scale.
    past : `bool`
        If `True` extend the range in the past.
    max_days : `int`
        Max. number of days for extension.
    """
    try:
        astro_time = astropy.time.Time(time, format="isot", scale="tai")
        max_delta = astropy.time.TimeDelta(float(max_days), format="jd", scale="tai")
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
 
    apdb = Apdb.from_uri(apdb_config)
    if not isinstance(apdb, ApdbCassandra):
        print("ERROR: Non-Cassandra APDB does not use time-partitioned tables.", file=sys.stderr)
        return 1
 
    admin = apdb.admin
    try:
        result = admin.extend_time_partitions(astro_time, forward=not past, max_delta=max_delta)
    except (TypeError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
 
    part_range = admin.time_partitions()
    start_time, _ = admin.partitioner.partition_period(part_range.start)
    _, end_time = admin.partitioner.partition_period(part_range.end)
 
    if result:
        print(
            "Time partition range succesfully extended.\n"
            f"New time partition range: {part_range.start} - {part_range.end} "
            f"[{start_time.tai.isot}, {end_time.tai.isot})"
        )
    else:
        print(
            "Time partition range was not extended.\n"
            f"Current time partition range: {part_range.start} - {part_range.end} "
            f"[{start_time.tai.isot}, {end_time.tai.isot})"
        )
 
    return 0
 
 

partition_show_temporal()

int lsst.dax.apdb.scripts.partition.partition_show_temporal

(

str

apdb_config

)

Print range of temporal partitions.

Parameters
----------
apdb_config : `str`
    URL for APDB configuration file.

Definition at line 35 of file partition.py.

def partition_show_temporal(apdb_config: str) -> int:
    """Print range of temporal partitions.
 
    Parameters
    ----------
    apdb_config : `str`
        URL for APDB configuration file.
    """
    apdb = Apdb.from_uri(apdb_config)
    if not isinstance(apdb, ApdbCassandra):
        print("ERROR: Non-Cassandra APDB does not use time-partitioned tables.", file=sys.stderr)
        return 1
 
    admin = apdb.admin
    try:
        part_range = admin.time_partitions()
    except TypeError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
 
    start_time, _ = admin.partitioner.partition_period(part_range.start)
    _, end_time = admin.partitioner.partition_period(part_range.end)
 
    print(
        f"Current time partition range: {part_range.start} - {part_range.end} "
        f"[{start_time.tai.isot}, {end_time.tai.isot})"
    )
 
    return 0