Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion Pipfile
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ pytest = "*"
ruff = "*"
setuptools = "*"
pip-audit = "*"
pytest-freezegun = "*"

[requires]
python_version = "3.12"
986 changes: 474 additions & 512 deletions Pipfile.lock

Large diffs are not rendered by default.

1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,6 @@ ignore = [
"PLR0912",
"PLR0913",
"PLR0915",
"S320",
"S321",
"S608",
"TRY003"
Expand Down
13 changes: 7 additions & 6 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,16 +182,16 @@ def dataset_with_same_day_runs(tmp_path) -> TIMDEXDataset:
# be represented.
run_params.extend(
[
(100, "alma", "2025-01-01", "full", "index", "run-1"),
(75, "alma", "2025-01-01", "full", "index", "run-2"),
(10, "alma", "2025-01-01", "daily", "index", "run-3"),
(20, "alma", "2025-01-02", "daily", "index", "run-4"),
(5, "alma", "2025-01-02", "daily", "delete", "run-5"),
(100, "alma", "2025-01-01", "full", "index", "run-1", "2025-01-01T01:00:00"),
(75, "alma", "2025-01-01", "full", "index", "run-2", "2025-01-01T02:00:00"),
(10, "alma", "2025-01-01", "daily", "index", "run-3", "2025-01-01T03:00:00"),
(20, "alma", "2025-01-02", "daily", "index", "run-4", "2025-01-02T01:00:00"),
(5, "alma", "2025-01-02", "daily", "delete", "run-5", "2025-01-02T02:00:00"),
]
)

for params in run_params:
num_records, source, run_date, run_type, action, run_id = params
num_records, source, run_date, run_type, action, run_id, run_timestamp = params
records = generate_sample_records(
num_records,
timdex_record_id_prefix=source,
Expand All @@ -200,6 +200,7 @@ def dataset_with_same_day_runs(tmp_path) -> TIMDEXDataset:
run_type=run_type,
action=action,
run_id=run_id,
run_timestamp=run_timestamp,
)
timdex_dataset.write(records)

Expand Down
46 changes: 2 additions & 44 deletions tests/test_dataset.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,13 @@
# ruff: noqa: D205, D209, S105, S106, SLF001, PD901, PLR2004
# ruff: noqa: D205, D209, SLF001, PLR2004

import os
from datetime import UTC, date, datetime
from datetime import date
from unittest.mock import MagicMock, patch

import pyarrow as pa
import pytest
from pyarrow import fs

from tests.utils import generate_sample_records
from timdex_dataset_api.dataset import (
DatasetNotLoadedError,
TIMDEXDataset,
Expand Down Expand Up @@ -466,47 +465,6 @@ def test_dataset_current_records_index_filtering_accurate_records_yielded(
]


@pytest.mark.freeze_time("2025-05-22 01:23:45.567890")
def test_dataset_write_includes_minted_run_timestamp(tmp_path):

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test no longer makes sense as TIMDEXDataset.write() no longer mints a timestamp if one is not provided.

Either run_timestamp is provided explicitly to each DatasetRecord for writing, or it defaults to an ISO timestamp version of run_date for that record (which is required).

# create dataset
location = str(tmp_path / "one_run_at_frozen_time")
os.mkdir(location)
timdex_dataset = TIMDEXDataset(location)

run_id = "abc123"

# perform a single ETL run that should pickup the frozen time for run_timestamp
records = generate_sample_records(
10,
timdex_record_id_prefix="alma",
source="alma",
run_date="2025-05-22",
run_type="full",
action="index",
run_id=run_id,
)
timdex_dataset.write(records)
timdex_dataset.load()

# assert TIMDEXDataset.write() applies current time as run_timestamp
run_row_dict = next(timdex_dataset.read_dicts_iter())
assert "run_timestamp" in run_row_dict
assert run_row_dict["run_timestamp"] == datetime(
2025,
5,
22,
1,
23,
45,
567890,
tzinfo=UTC,
)

# assert the same run_timestamp is applied to all rows in the run
df = timdex_dataset.read_dataframe(run_id=run_id)
assert len(list(df.run_timestamp.unique())) == 1


def test_dataset_load_current_records_gets_correct_same_day_full_run(
dataset_with_same_day_runs,
):
Expand Down
2 changes: 1 addition & 1 deletion tests/test_read.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# ruff: noqa: PLR2004, PD901
# ruff: noqa: PLR2004

import pandas as pd
import pyarrow as pa
Expand Down
65 changes: 64 additions & 1 deletion tests/test_records.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import re
from datetime import date
from datetime import UTC, date, datetime

import pytest

Expand Down Expand Up @@ -71,7 +71,70 @@ def test_dataset_record_serialization():
"action": "index",
"run_id": "abc123",
"run_record_offset": 0,
"run_timestamp": datetime(2024, 12, 1, 0, 0, tzinfo=UTC),
"year": "2024",
"month": "12",
"day": "01",
}


@pytest.mark.parametrize(
("run_timestamp_input", "expected_run_timestamp", "expected_exception"),
[
(
None,
None,
TypeError, # expecting string, not None
),
(
date(2025, 1, 1),
None,
TypeError, # expecting string, not datetime object
),
(
"2024-12-01T10:00:00Z",
datetime(2024, 12, 1, 10, 0, tzinfo=UTC),
None,
),
(
"2024-12-01T23:59:59.999999+00:00",
datetime(2024, 12, 1, 23, 59, 59, 999999, tzinfo=UTC),
None,
),
],
)
def test_dataset_record_run_timestamp_parsing(
run_timestamp_input, expected_run_timestamp, expected_exception
):
values = {
"timdex_record_id": "alma:123",
"source_record": b"<record><title>Hello World.</title></record>",
"transformed_record": b"""{"title":["Hello World."]}""",
"source": "libguides",
"run_date": "2024-12-01",
"run_type": "full",
"action": "index",
"run_id": "abc123",
"run_record_offset": 0,
"run_timestamp": run_timestamp_input,
}
if not expected_exception:
dataset_record = DatasetRecord(**values)
assert dataset_record.to_dict() == {
"timdex_record_id": "alma:123",
"source_record": b"<record><title>Hello World.</title></record>",
"transformed_record": b"""{"title":["Hello World."]}""",
"source": "libguides",
"run_date": date(2024, 12, 1),
"run_type": "full",
"action": "index",
"run_id": "abc123",
"run_record_offset": 0,
"run_timestamp": expected_run_timestamp,
"year": "2024",
"month": "12",
"day": "01",
}
else:
with pytest.raises(expected_exception):
DatasetRecord(**values)
2 changes: 1 addition & 1 deletion tests/test_write.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# ruff: noqa: S105, S106, SLF001, PLR2004, PD901, D209, D205
# ruff: noqa: PLR2004, D209, D205
import math
import os
from unittest.mock import patch
Expand Down
2 changes: 2 additions & 0 deletions tests/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ def generate_sample_records(
run_type: str | None = "daily",
action: str | None = "index",
run_id: str | None = None,
run_timestamp: str | None = None,
) -> Iterator[DatasetRecord]:
"""Generate sample DatasetRecords."""
if not run_id:
Expand All @@ -33,6 +34,7 @@ def generate_sample_records(
action=action,
run_id=run_id,
run_record_offset=x,
run_timestamp=run_timestamp or run_date,
)


Expand Down
2 changes: 1 addition & 1 deletion timdex_dataset_api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from timdex_dataset_api.dataset import TIMDEXDataset
from timdex_dataset_api.record import DatasetRecord

__version__ = "2.0.0"
__version__ = "2.1.0"

__all__ = [
"DatasetRecord",
Expand Down
9 changes: 1 addition & 8 deletions timdex_dataset_api/dataset.py
Original file line number Diff line number Diff line change
Expand Up @@ -423,17 +423,10 @@ def create_record_batches(
Args:
- records_iter: Iterator of DatasetRecord instances
"""
run_timestamp = datetime.now(UTC)
for i, record_batch in enumerate(
itertools.batched(records_iter, self.config.write_batch_size)
):
record_dicts = [
{
**record.to_dict(),
"run_timestamp": run_timestamp,
}
for record in record_batch
]
Comment on lines -426 to -436

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is really the most important part of the PR: we no longer mint a run_timestamp value in this method, but instead assume/require that all DatasetRecord's that are getting written will have a value already.

record_dicts = [record.to_dict() for record in record_batch]
batch = pa.RecordBatch.from_pylist(record_dicts)
logger.debug(f"Yielding batch {i + 1} for dataset writing.")
yield batch
Expand Down
17 changes: 17 additions & 0 deletions timdex_dataset_api/record.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,23 @@

from datetime import UTC, date, datetime

import attrs
from attrs import asdict, define, field


def strict_date_parse(date_string: str) -> date:
return datetime.strptime(date_string, "%Y-%m-%d").astimezone(UTC).date()


def datetime_iso_parse(datetime_iso_string: str) -> datetime:
parsed_datetime = datetime.fromisoformat(datetime_iso_string)
# if timezone not present, set as UTC and return
if parsed_datetime.tzinfo is None:
return parsed_datetime.replace(tzinfo=UTC)
# else, convert to / ensure UTC and return
return parsed_datetime.astimezone(UTC)


@define
class DatasetRecord:
"""Container for single dataset record.
Expand All @@ -26,6 +36,13 @@ class DatasetRecord:
run_type: str = field()
action: str = field()
run_id: str = field()
run_timestamp: datetime = field(
converter=datetime_iso_parse,
default=attrs.Factory(
lambda self: self.run_date.isoformat(),
takes_self=True,
),
)
run_record_offset: int = field(default=None)

@property
Expand Down