ccat_ops_db package

Subpackages

ccat_ops_db.config package

Submodules

ccat_ops_db.ccat_ops_db module

Generate database URL based on type and configuration.

Parameters:

database_type – Type of database (‘sqlite’, ‘mysql’, ‘postgresql’)
database – Optional database URL override
host – Optional host override
port – Optional port override
user – Optional user override
password – Optional password override
db_name – Optional database name override
async_driver – Whether to use async driver (e.g., asyncpg for PostgreSQL)

Initialize a database connection based on the supplied configuration.

Parameters:

database_type: str, optional: Can be ‘sqlite’, ‘mysql’, or ‘postgresql’. Defaults to config setting.
database: str, optional: URL string to connect to the database. Defaults to config setting.
drop: bool, default False: If True, drops all tables before creating them.
null_pool: bool, default False: If True, uses NullPool instead of the default connection pool.
host: str, optional: Database host override.
port: str, optional: Database port override.
user: str, optional: Database user override.
password: str, optional: Database password override.
db_name: str, optional: Database name override.

Returns:

Tuple[scoped_session, Engine]: A tuple containing the database session and engine.

ccat_ops_db.init_obs_units module

ccat_ops_db.init_obs_units.clean_line(line)

ccat_ops_db.init_obs_units.extract_from_json_list(tbl, obsunit_name)

Extract a json string for a given obsunit

Parameters:

tbl (list of json) – Each json string has the name of obsunit and parameters. For example, If it is table_azimuth, the combination of elevation, rising/setting, az_min/max and duration as a list If it is table_mapping_params, then the mapping parameters.
obsunit_name (str) – Name of the obsunit to be searched in the json string list

Returns:

extracted – A json string extracted from tbl that specifies the parameters of a given obsunit

Return type:

json

ccat_ops_db.init_obs_units.init_obs_units(installation_path, input_yaml_file, session)

ccat_ops_db.init_obs_units.match_kosma_source_patterns(source_string)

ccat_ops_db.init_obs_units.read_inpar_parameters(inpar_parameters_file)

reads the inpar parameter file

This file is a text file with the following format:

# This is a comment KEYWORD1=VALUE1 KEYWORD2=VALUE2

There can be any amount of spaces around the = sign Empty lines are ignored

ccat_ops_db.init_obs_units.read_obs_units(obsunits_list, obs_unit_type='chai_obs_unit', inpar_parameters_folder=None, tiling_parameters_folder=None, table_azimuth_file=None, table_mapping_params_file=None, session=None): reads csv file with first line as header and returns a list of dictionaries

ccat_ops_db.init_obs_units.read_pre_scheduled_slots(pre_scheduled_file, session)

Read pre-scheduled slots and add them to the database session

Parameters:

pre_scheduled_file (str) – Input file name with pre-scheduled obsunits and time
session (database session)

ccat_ops_db.init_obs_units.read_sources(source_file, file_format, source_type, session)

Read a source file and add sources to the database session

Parameters:

source_file (str) – Input file name with source namd and coordinates
file_format (str) – kosma (kosma_software source file format) or csv. If csv, the coordinate system is assumed to be icrs.
source_type (str) – None is fixed target with coordinates. Other names should be compatible with the object in models.
session (database session)

ccat_ops_db.init_obs_units.read_tiling_parameters(tiling_parameters_file)

reads the tiling parameter file

The format of the file is tile_iD / tile_offset_x / tile_offset_y / x_or_y / tile_unit_scaling_x / tile_unit_scaling_y / edge / inpar_file / goal_ncycle / status_ncycle

lines starting with # are comments empty lines are ignored priority_in_tiling is the order in the file

ccat_ops_db.models module

class ccat_ops_db.models.ApiToken(**kwargs)

Bases: Base

API tokens for programmatic access to the API

active: Whether token is active

created_at

expires_at: Token expiration time

id

is_expired(): Check if token is expired

is_valid(): Check if token is valid (active and not expired)

last_used_at: Last time token was used

last_used_ip: IP address of last usage

name: Human-readable name for the token

scopes: List of permission scopes for this token

token_hash: Hashed token value

token_prefix: First few characters for identification

usage_count: Number of times token was used

user

user_id

class ccat_ops_db.models.ChaiInparParameter(**kwargs)

Bases: Base

act_pixflg: Flag to use the actual pixel position of refpoint. Default = 0

bet: Map center offset in latitude [arcsec]

chai_tilings

cormap: Map cood. system

corref: Reference cood. system

crosssizeb: Size of cross in latitude [arcsec]

crosssizel: Size of cross in longitude [arcsec]

evendump: If 1 allow even number of dump positions without hitting the center. Default = 0

goal_resolution: Spectral resolution [km/s] with which the scientific analysis will be done

history

id

lam: Map center offset in longitude [arcsec]

line_range

v,-10:10:v”)

Type:: Range of velocities [km/s] affected by lines (e.g. “-70
Type:: -30

lineperoff: Define how many otf scan lines for one off measurement

mapangle: Position angle of the map (counter-clock) [degree]

mapsizeb: Size of map in latitude [arcsec]

mapsizel: Size of map in longitude [arcsec]

mode: Observing mode; otfl or otfb

name

nextflg: If Y, enable telescope moving when writing data. Default = N

nmapb: number of raster positions along latitude

nmapl: number of raster positions along longitude

novertical: If 1 skip the second (vertical) scan for cross observations. Default = 0

offononoff: Sequence of OFF and ON (1 = OFF-ON-ON-OFF (default), 0 = OFF-ON)

offperload: Define how many off for one load measurment

offperpattern: Define how many off measurement per pattern

onperload: Define how many on positions for one load measurement

onperoff: Define how many on positions for one off measurement

otfpattern: OTF pattern file name specified in kosma_software

pointingflg: Flag to indicate that it is a pointing session. Default = 0

refname: Reference name when using the absolute positions. Set NAN to use relative

refoffb: Reference position relative to on in latitude [arcesc]

refoffl: Reference position relative to on in longitude [arcsec]

refpoint: Reference point (as is used in setpoint)

repetition: Repetition number

repperload: Define how many repetition for one load measurment

reverseflg

0 = (+x,+y), 1 = (-x,-y), 2 = (-x,+x,-y,+y)

Type:: OTF scan direction in cross

scan_dir: OTF scan direction (1 is +x or +y, -1 is -x or -y)

scan_order: Order of OTF lines (1 is +x or +y, -1 is -x or -y)

stepb: Step size in latitude [arcsec]

stepl: Step size in longitude [arcsec]

toff: Forced off time [s] (-1 to use the default calculation = default)

ton: On integration time [sec]

version

class ccat_ops_db.models.ChaiModuleConfiguration(**kwargs)

Bases: InstrumentModuleConfiguration

config_parameters: List of instrument configuration parameters

id

instrument_module

instrument_module_id

line

line_id

raw_data_files

type

class ccat_ops_db.models.ChaiObservationConfiguration(**kwargs)

Bases: ObservationConfiguration

azimuth_range: Azimuth range lookup table for constant elevation scans

chai_tilings

id

ntilelines: Number of tile lines to be grouped (for socring in scheduler)

obs_units

type

class ccat_ops_db.models.ChaiTiling(**kwargs)

Bases: Base

chai_inpar_parameter

chai_inpar_parameter_id

chai_observation_configuration

chai_observation_configuration_id

edge

goal_ncycle

history

id

priority_in_tiling

tile_id

tile_offset_x

tile_offset_y

tile_unit_scaling_x

tile_unit_scaling_y

version

x_or_y

class ccat_ops_db.models.ConstantElevationSource(**kwargs)

Bases: Source

A source that is observed at a constant elevation

This class is a subclass of the Source class and inherits all attributes from the Source class.

dec_deg_max: The maximum declination of the area in degrees

dec_deg_min: The minimum declination of the area in degrees

frame: The frame of the coordinates of the original input

id

name: The name of the source

obs_units

ra_deg_max: The maximum right ascension of the area in degrees

ra_deg_min: The minimum right ascension of the area in degrees

sbet_max: The maximum latitude string of the original input

sbet_min: The minimum latitude string of the original input

slam_max: The maximum longitude string of the original input

slam_min: The minimum longitude string of the original input

type: The polymorphic identity

vlsr: The local standard of rest velocity in km/s

class ccat_ops_db.models.DataArchive(**kwargs)

Bases: Base

id

name

class ccat_ops_db.models.DataLocation(**kwargs)

Bases: Base

Base class for all data storage locations with polymorphic storage types.

active

id

location_type

name

priority

site

site_id

storage_type

class ccat_ops_db.models.DataTransfer(**kwargs)

Bases: Base

Tracks data transfers between locations.

data_transfer_method

data_transfer_package

data_transfer_package_id

destination_location

destination_location_id

end_time

failure_error_message

id

logs

origin_location

origin_location_id

process_id

retry_count

start_time

status

transfer_program_log

unpack_end_time

unpack_failure_error_message

unpack_log

unpack_retry_count

unpack_start_time

unpack_status

class ccat_ops_db.models.DataTransferLog(**kwargs)

Bases: SystemLog

Log entries for data transfers with references to log files.

This model implements a lightweight approach where: - Basic status info and log file path are stored in the database - Full command outputs are stored in files - Detailed metrics are stored in InfluxDB

property content: str | None: Read and return the full log content if file exists.

data_transfer

data_transfer_id

id

log_path: Path to file containing full command output

status: Status of this transfer attempt (success/failure)

timestamp

type

class ccat_ops_db.models.DataTransferPackage(**kwargs)

Bases: Base

checksum

data_transfers

failure_error_message

file_name

hash_id

id

origin_location

origin_location_id

physical_copies

raw_data_files

raw_data_packages

relative_path

retry_count

size

status

class ccat_ops_db.models.DataTransferPackagePhysicalCopy(**kwargs)

Bases: PhysicalCopy

Tracks physical copies of data transfer packages.

checksum

created_at

data_location

data_location_id

data_transfer_package

data_transfer_package_id

deleted_at

deletion_task_id

property full_path: Get the full path/S3 key for this physical copy.

id

status

type

verified_at

class ccat_ops_db.models.DataTransferRoute(**kwargs)

Bases: Base

Defines routes for data transfer between sites and locations.

The database implements a flexible routing system that supports: 1. Direct routes between specific locations 2. Relay routes through intermediate sites 3. Custom location-to-location overrides

destination_location

destination_location_id

destination_site

destination_site_id

id

name

origin_location

origin_location_id

origin_site

origin_site_id

relay_site

relay_site_id

route_type

transfer_method

class ccat_ops_db.models.DiskDataLocation(**kwargs)

Bases: DataLocation

Disk-based storage location.

active

host

id

location_type

name

path

priority

site

site_id

storage_type

user

class ccat_ops_db.models.ExecutedObsUnit(**kwargs)

Bases: Base

end_time

id

mean_elevation

mean_pwv

obs_info: Static ancillary infomation (e.g. ObsUnit version, tiling ID for CHAI)

obs_progress: Field for progress tracking

obs_unit

obs_unit_id

raw_data_files

raw_data_packages

start_time

status: running/success/interruped/unusable/…

class ccat_ops_db.models.FixedSource(**kwargs)

Bases: Source

A source that has a fixed coordinates

This class is a subclass of the Source class and inherits all attributes from the Source class.

The class implements the skycoord property that returns a SkyCoord object from the ra_deg and dec_deg columns.

dec_deg: The declination in degrees (ICRS)

frame: The frame of the coordinates of the original input

id

name: The name of the source

obs_units

ra_deg: The right ascension in degrees (ICRS)

sbet: Latitude string of the original input

property skycoord: Return a SkyCoord object from the ra_deg and dec_deg columns

slam: Longitude string of the original input

type: The polymorphic identity

vlsr: The local standard of rest velocity in km/s

class ccat_ops_db.models.Instrument(**kwargs)

Bases: Base

available

description

id

instrument_type

modules

name

observing_programs

telescope

telescope_id

class ccat_ops_db.models.InstrumentModule(**kwargs)

Bases: Base

available

description

id

instrument

instrument_id

instrument_module_configurations

name

pixels

raw_data_files

raw_data_packages

class ccat_ops_db.models.InstrumentModuleConfiguration(**kwargs)

Bases: Base

id

instrument_module

instrument_module_id

raw_data_files

type

class ccat_ops_db.models.Line(**kwargs)

Bases: Base

A spectral line that is observed by an instrument

available

comment

id

name: The name of the spectral line

rest_frequency: The rest frequency of the line

side_band: The side band of the line

class ccat_ops_db.models.LocationType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

BUFFER = 'buffer'

LONG_TERM_ARCHIVE = 'long_term_archive'

PROCESSING = 'processing'

SOURCE = 'source'

class ccat_ops_db.models.LongTermArchiveTransfer(**kwargs)

Bases: Base

attempt_count

destination_data_location

destination_data_location_id

end_time

error_message

failure_error_message

id

last_attempt_time

logs

origin_data_location

origin_data_location_id

raw_data_package

raw_data_package_id

requires_intervention

site

site_id

start_time

status

class ccat_ops_db.models.LongTermArchiveTransferLog(**kwargs)

Bases: SystemLog

id

log

long_term_archive_transfer

long_term_archive_transfer_id

timestamp

type

class ccat_ops_db.models.ObsMode(**kwargs)

Bases: Base

description

id

name

obs_units

class ccat_ops_db.models.ObsUnit(**kwargs)

Bases: Base

additional_parameters: Any special parameters or constraints to be stored

available: Whether this ObsUnit is ready to be scheduled

cadence

equal_tolerance: Tolerance of inbalanced schedule when group_type is equal

executed_obs_units

group

group_type

history

id

instrument_module_configurations

lsa_margin

max_alt

max_distance_in_map

max_lsa

max_rotang

min_alt

min_lsa

min_rotang

name

nominal_alt

obs_mode

obs_mode_id

observation_configuration

observation_configuration_id

observing_program

observing_program_id

phase: CM=commissioning BS=baseline science etc.

pre_scheduled_basis: Whether this Obsunit is executed on pre-scheduled basis, meaning that it has a very low priority outside of the pre-scheduled slots.

pre_scheduled_slots

primary_instrument_module_configuration

primary_instrument_module_configuration_id

priorities

raw_data_files

raw_data_packages

requested_time_h

source

source_id

sub_observing_program

sub_observing_program_id

trans_ref: Reference transmission

unit_duration_h

version

class ccat_ops_db.models.ObservationConfiguration(**kwargs)

Bases: Base

azimuth_range: Azimuth range lookup table for constant elevation scans

id

obs_units

type

class ccat_ops_db.models.Observatory(**kwargs)

Bases: Base

description

id

name

telescopes

class ccat_ops_db.models.ObservingProgram(**kwargs)

Bases: Base

description

id

instruments

lead

lead_id

name

obs_units

short_name

sub_observing_programs

class ccat_ops_db.models.PackageState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ARCHIVED = 'archived'

FAILED = 'failed'

TRANSFERRING = 'transferring'

WAITING = 'waiting'

class ccat_ops_db.models.PhysicalCopy(**kwargs)

Bases: Base

Base class for tracking physical copies of files across different storage locations.

This class implements a polymorphic pattern to track physical copies of different types of files (RawDataFile, RawDataPackage, DataTransferPackage) across different storage locations and types. Each physical copy represents an actual file on disk or other storage medium.

checksum

created_at

data_location

data_location_id

deleted_at

deletion_task_id

id

status

type

verified_at

class ccat_ops_db.models.PhysicalCopyStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

DELETED = 'deleted'

DELETION_FAILED = 'deletion_failed'

DELETION_IN_PROGRESS = 'deletion_in_progress'

DELETION_PENDING = 'deletion_pending'

DELETION_POSSIBLE = 'deletion_possible'

DELETION_SCHEDULED = 'deletion_scheduled'

PRESENT = 'present'

STAGED = 'staged'

class ccat_ops_db.models.PreScheduledSlot(**kwargs)

Bases: Base

end_time

id

obs_unit

obs_unit_id

start_time

class ccat_ops_db.models.PrimeCamModuleConfiguration(**kwargs)

Bases: InstrumentModuleConfiguration

config_parameters: List of instrument configuration parameters

id

instrument_module

instrument_module_id

raw_data_files

type

class ccat_ops_db.models.PrimeCamObservationConfiguration(**kwargs)

Bases: ObservationConfiguration

azimuth_range: Azimuth range lookup table for constant elevation scans

history

id

mapping_parameters: List of mapping parameters

obs_units

type

version

class ccat_ops_db.models.RawDataFile(**kwargs)

Bases: Base

Represents a raw data file from an instrument.

checksum

created_at

data_transfer_package

data_transfer_package_id

description

executed_obs_unit

executed_obs_unit_id

file_type

id

instrument_module

instrument_module_configuration

instrument_module_configuration_id

instrument_module_id

name

obs_unit

obs_unit_id

physical_copies

raw_data_package

raw_data_package_id

relative_path

size

source_location

source_location_id

state

class ccat_ops_db.models.RawDataFilePhysicalCopy(**kwargs)

Bases: PhysicalCopy

Tracks physical copies of individual raw data files.

checksum

created_at

data_location

data_location_id

deleted_at

deletion_task_id

property full_path: Get the full path/S3 key for this physical copy.

id

raw_data_file

raw_data_file_id

status

type

verified_at

class ccat_ops_db.models.RawDataPackage(**kwargs)

Bases: Base

A raw data package is a bundle of raw data files that were observed in an observation unit. But they should never be larger than 50GB in size.

analyze_status

checksum

created_at

data_transfer_package

data_transfer_package_id

executed_obs_unit

executed_obs_unit_id

id

instrument_module

instrument_module_id

long_term_archive_transfers

name

obs_unit

obs_unit_id

package_metadata

physical_copies

raw_data_files

relative_path: This is the relative path to the raw data package. The absolute path of the location of this file for each archive is stored in the DataLocation table. The path is relative to the raw_data_path of the DataLocation.

retry_count

size

staging_jobs

state

status

class ccat_ops_db.models.RawDataPackageMetadata(**kwargs)

Bases: Base

Model for storing additional metadata for raw data packages.

This table stores metadata that is not part of the core database models but is needed for IVOA-compatible metadata generation. It includes: - Instrument-specific parameters - Additional quality metrics - Extended provenance information - Custom metadata fields

created_at

custom_metadata

id

instrument_specific

provenance

quality_metrics

raw_data_package

raw_data_package_id

updated_at

class ccat_ops_db.models.RawDataPackagePhysicalCopy(**kwargs)

Bases: PhysicalCopy

Tracks physical copies of raw data packages.

checksum

created_at

data_location

data_location_id

deleted_at

deletion_task_id

property full_path: Get the full path/S3 key for this physical copy.

id

raw_data_package

raw_data_package_id

status

type

verified_at

class ccat_ops_db.models.Role(**kwargs)

Bases: Base

description

github_team_mappings: GitHub teams that map to this role

id

name

permissions: List of permissions granted to this role

users

class ccat_ops_db.models.RouteType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

CUSTOM = 'custom'

DIRECT = 'direct'

RELAY = 'relay'

class ccat_ops_db.models.S3DataLocation(**kwargs)

Bases: DataLocation

S3-compatible object storage location.

active

bucket_name

endpoint_url

get_s3_credentials(site_name: str) → tuple[str, str]

Get S3 credentials from environment variables using the pattern: {Site.name}_{DataLocation.name}_S3_ACCESS_KEY_ID {Site.name}_{DataLocation.name}_S3_SECRET_ACCESS_KEY

Examples: - Cologne site, long_term_archive location:

COLOGNE_LONG_TERM_ARCHIVE_S3_ACCESS_KEY_ID=your_access_key COLOGNE_LONG_TERM_ARCHIVE_S3_SECRET_ACCESS_KEY=your_secret_key

US site, long_term_archive location: US_LONG_TERM_ARCHIVE_S3_ACCESS_KEY_ID=your_access_key US_LONG_TERM_ARCHIVE_S3_SECRET_ACCESS_KEY=your_secret_key

Parameters:: site_name (str) – Name of the site
Returns:: (access_key_id, secret_access_key)
Return type:: tuple[str, str]

id

location_type

name

priority

region

site

site_id

storage_type

class ccat_ops_db.models.Site(**kwargs)

Bases: Base

Represents a physical or logical site where data can be stored or processed.

id

locations

long_term_archive_transfers

name

short_name

site_location

class ccat_ops_db.models.SolarSystemObject(**kwargs)

Bases: Source

A source that is a solar system object

This class is a subclass of the Source class and inherits all attributes from the Source class.

eph_name: Standard ephemeris name of the source

id

naif_id: The NAIF ID of the source

name: The name of the source

obs_units

type: The polymorphic identity

class ccat_ops_db.models.Source(**kwargs)

Bases: Base

A source is a celestial object

This class serves as a base class for various types of sources. It is a polymorphic class in SQLAlchemy, not instantiated directly, but used to provide common attributes. Subclasses, implemented as separate database tables, inherit from the source class and can have additional specific attributes.

See classes that are based on this class for more information on the implemented types of sources.

id

name: The name of the source

obs_units

type: The polymorphic identity

class ccat_ops_db.models.StagingJob(**kwargs)

Bases: Base

active

destination_data_location

destination_data_location_id

end_time

failure_error_message

id

logs

origin_data_location

origin_data_location_id

raw_data_packages

retry_count

start_time

status

class ccat_ops_db.models.StagingJobLog(**kwargs)

Bases: SystemLog

id

log

staging_job

staging_job_id

timestamp

type

class ccat_ops_db.models.Status(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

COMPLETED = 'completed'

FAILED = 'failed'

IN_PROGRESS = 'in_progress'

PENDING = 'pending'

SCHEDULED = 'scheduled'

class ccat_ops_db.models.StorageType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

DISK = 'disk'

S3 = 's3'

TAPE = 'tape'

class ccat_ops_db.models.SubObservingProgram(**kwargs)

Bases: Base

description

id

name

obs_units

observing_program

observing_program_id

short_name

class ccat_ops_db.models.SystemLog(**kwargs)

Bases: Base

id

type

class ccat_ops_db.models.TapeDataLocation(**kwargs)

Bases: DataLocation

Tape-based storage location.

active

id

library_name

location_type

mount_path

name

priority

site

site_id

storage_type

class ccat_ops_db.models.Telescope(**kwargs)

Bases: Base

alt_m

description

id

instruments

lat_deg

lon_deg

name

observatory

observatory_id

class ccat_ops_db.models.User(**kwargs)

Bases: Base

active

affiliation

api_tokens

confirmed_at

current_login_at

current_login_ip

email

first_name

github_id: GitHub user ID

github_username: GitHub username

id

last_login_at

last_login_ip

last_name

login_count

observing_programs

password

preferences: User preferences and settings

roles

title

username

ccat_ops_db.opsdb_init module

ccat_ops_db.opsdb_init.add_or_update_data_transfer_routes(session, values)

ccat_ops_db.opsdb_init.add_or_update_instrument(session, values)

ccat_ops_db.opsdb_init.add_or_update_line(session, key, values)

ccat_ops_db.opsdb_init.add_or_update_module(session, values, instrument)

ccat_ops_db.opsdb_init.add_or_update_obs_mode(session, key, values)

ccat_ops_db.opsdb_init.add_or_update_observatory(session, values)

ccat_ops_db.opsdb_init.add_or_update_observing_program(session, values)

ccat_ops_db.opsdb_init.add_or_update_role(session, values)

ccat_ops_db.opsdb_init.add_or_update_site_and_locations(session, site_key, values)

Add or update a site and its associated data locations.

Parameters:

session – SQLAlchemy session
site_key – The key from the TOML file (e.g. “CologneStaging”)
values – Dictionary containing site and location information

ccat_ops_db.opsdb_init.add_or_update_sub_observing_program(session, values, observing_program)

ccat_ops_db.opsdb_init.add_or_update_telescope(session, values)

ccat_ops_db.opsdb_init.add_or_update_user(session, values)

ccat_ops_db.opsdb_init.check_successfull_initialization()

ccat_ops_db.opsdb_init.get_toml_files_in_folder(folder)

This function takes a folder and looks for all .toml files in it returning a list of them

Args:FastAPI: folder (str): the folder to look for .toml files in

Returns:: a list of all the .toml files in the folder
Return type:: list

ccat_ops_db.opsdb_init.opsdb_initialization(verbose=False, drop=False, local_url=None, docker_compose=False, database=None, data_archive_mode='development'): This function provisions the ccat_obs_db for testing purposes

ccat_ops_db.opsdb_init.successfull_initialization()

ccat_ops_db.opsdb_init.unique_id()

ccat_ops_db.render_database_schema module

ccat_ops_db.render_database_schema.render_database_schema(file_type='png', file_name='ccat_ops_db_schema.png', output_path='/tmp/')

ccat_ops_db.render_database_schema.upload_to_dokuwiki(dokuwiki_path, local_path)

Here we use the python dokuwiki api to upload the newly created

Parameters:

dokuwiki_path (str) – Dokuwiki path name: NAMESPACE:IMAGE_PATH
local_path (str) – Local path to the file

ccat_ops_db.utils module

ccat_ops_db.utils.add_inpar_and_tiling(session, observation_configuration, inpar_parameters_list=None, tiling_parameters=None)

Add inpar and tiling parameters to the observation_configuration

Parameters:

session (database session)
observation_configuration (ccat.opsdb.observation_configuration) – Observation configuration with inpar and tiling to be associated
inpar_parameters_list (list) – List of dictionary of inpar_parameters
tiling_parameters (list) – List of dictionary of each line in tiling

Returns:

observation_configuration – Updated observation configuration

Return type:

ccat.opsdb.observation_configuration

ccat_ops_db.utils.add_or_update_obs_unit(session, values, obs_unit_type='chai_obs_unit', inpar_parameters_list=None, tiling_parameters=None, instrument_module_configurations=None, azimuth_range=None, mapping_parameters=None)

ccat_ops_db.utils.add_or_update_source(session, values, source_type=None)

ccat_ops_db.utils.add_pre_scheduled_slot(session, values)

Module contents

Generate database URL based on type and configuration.

Parameters:

database_type – Type of database (‘sqlite’, ‘mysql’, ‘postgresql’)
database – Optional database URL override
host – Optional host override
port – Optional port override
user – Optional user override
password – Optional password override
db_name – Optional database name override
async_driver – Whether to use async driver (e.g., asyncpg for PostgreSQL)

Initialize a database connection based on the supplied configuration.

Parameters:

database_type: str, optional: Can be ‘sqlite’, ‘mysql’, or ‘postgresql’. Defaults to config setting.
database: str, optional: URL string to connect to the database. Defaults to config setting.
drop: bool, default False: If True, drops all tables before creating them.
null_pool: bool, default False: If True, uses NullPool instead of the default connection pool.
host: str, optional: Database host override.
port: str, optional: Database port override.
user: str, optional: Database user override.
password: str, optional: Database password override.
db_name: str, optional: Database name override.

Returns:

Tuple[scoped_session, Engine]: A tuple containing the database session and engine.