Conversion from data tables to NtpMeasurement class

server.app.db.db_interaction.row_to_dict(m, t)[source]

Converts a Measurement and Time SQLAlchemy row into a dictionary.

Parameters:

  • m (Measurement) – The measurement row containing NTP measurement data.

  • t (Time) – The time row containing timestamp data for the measurement.

Returns:

A dictionary representation of the combined measurement and timestamp data.

Return type:

dict[str, Any]

server.app.db.db_interaction.rows_to_dicts(rows)[source]

Converts a list of Measurement-Time row tuples into a list of dictionaries.

Parameters:

rows (list[Row[tuple[Measurement, Time]]]) – List of database rows containing Measurement and Time.

Returns:

A list of dictionaries where each dictionary contains combined data from Measurement and Time.

Return type:

list[dict[str, Any]]

server.app.db.db_interaction.dict_to_measurement(entry)[source]

Converts a dictionary representation of a measurement into an NtpMeasurement object.

Parameters:

entry (dict[str, Any]) – A dictionary containing the keys needed to construct an NtpMeasurement object.

Returns:

A fully constructed NtpMeasurement using the provided data.

Return type:

NtpMeasurement

Raises:

InvalidMeasurementDataError – If required keys are missing or construction fails due to invalid types/values.

server.app.db.db_interaction.rows_to_measurements(rows)[source]

Converts a list of Measurement-Time row tuples into NtpMeasurement objects.

Parameters:

rows (list[Row[tuple[Measurement, Time]]]) – List of database rows containing Measurement and Time data.

Returns:

A list of NtpMeasurement objects created from the row data.

Return type:

list[NtpMeasurement]

Methods used for interaction with the database

Insertion

server.app.db.db_interaction.insert_measurement(measurement, session)[source]

Inserts a new NTP measurement into the database. Before inserting, it sanitizes the string fields, because some fields may have a null character at the end which should be removed.

This function stores both the raw timestamps (in the times table) and the processed measurement data (in the measurements table). It wraps operations in a single transaction to ensure consistency and atomicity. If any insert fails, the transaction is rolled back.

Parameters:

  • measurement (NtpMeasurement) – The measurement data to store.

  • session (Session) – The currently active database session.

Raises:

DatabaseInsertError – If inserting the measurement or timestamps fails.

Return type:

None

Notes

  • Timestamps are stored with both second and fractional parts.

  • A foreign key (time_id) is used to link measurements to the times table.

  • Any failure within the transaction block results in automatic rollback.

Fetching data based on IP

server.app.db.db_interaction.get_measurements_timestamps_ip(session, ip, start, end)[source]

Fetch measurements for a specific IP address within a precise time range.

This function queries the measurements table, joined with the times table, and filters the results by: - The NTP server IP (ntp_server_ip) - The timestamp range (client_sent field) between start and end

Parameters:

  • session (Session) – The currently active database session.

  • ip (IPv4Address | IPv6Address | None) – The IP address of the NTP server.

  • start (PreciseTime) – The start of the time range to filter on.

  • end (PreciseTime) – The end of the time range to filter on.

Returns:

A list of measurement records. Each record includes:

  • IP, version, stratum

  • client/server send/receive timestamps with fractional parts

  • other measurement metadata

Return type:

list[dict]

Raises:

MeasurementQueryError – If the database query fails.

Fetching data based on domain name

server.app.db.db_interaction.get_measurements_timestamps_dn(session, dn, start, end)[source]

Fetches measurements for a specific domain name within a precise time range.

Similar to get_measurements_timestamps_ip, but filters by ntp_server_name. instead of ntp_server_ip.

Parameters:

  • session (Session) – The currently active database session.

  • dn (str) – The domain name of the NTP server.

  • start (PreciseTime) – The start of the time range to filter on.

  • end (PreciseTime) – The end of the time range to filter on.

Returns:

A list of measurement records (as dictionaries), each including:

  • Measurement metadata (domain name, version, etc.)

  • Timing data (client/server send/receive with precision)

Return type:

list[dict]

Raises:

MeasurementQueryError – If the database query fails.

Fetching measurements for jitter calculation

server.app.db.db_interaction.get_measurements_for_jitter_ip(session, ip, number=7)[source]

Fetches the last specified number (default 7) of measurements for specific IP address for calculating the jitter.

This function queries the measurements table, joined with the times table, and filters the results by: The NTP server IP (ntp_server_ip) and limits the result to the number specified.

Parameters:

  • session (Session) – The currently active database session.

  • ip (IPv4Address | IPv6Address) – The IP address of the NTP server.

  • number (int) – The number of measurements to get.

Returns:

A list of measurement records (as dictionaries), each including

  • Measurement metadata (IP, version, stratum, etc.)

  • Timing data (client/server send/receive with fractions)

Return type:

list[dict]

Raises:

MeasurementQueryError – If the database query fails.