Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: gracefully continue on tempfile cleanup failure #103

Merged
merged 3 commits into from
Dec 4, 2024
Merged

fix: gracefully continue on tempfile cleanup failure #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fbf007a
fix: gracefully continue on tempfile cleanup failure
aaronsteers Dec 3, 2024
a251049
drive-by mypy fixes
aaronsteers Dec 3, 2024
67ac101
Merge branch 'main' into aj/fix/gracefully-recover-from-tempfile-clea…
aaronsteers Dec 4, 2024
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
14 changes: 12 additions & 2 deletions airbyte_cdk/entrypoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@
from typing import Any, DefaultDict, Iterable, List, Mapping, Optional
from urllib.parse import urlparse

import orjson
import requests
from orjson import orjson
from requests import PreparedRequest, Response, Session

from airbyte_cdk.connector import TConfig
Expand Down Expand Up @@ -129,7 +129,11 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:

source_spec: ConnectorSpecification = self.source.spec(self.logger)
try:
with tempfile.TemporaryDirectory() as temp_dir:
with tempfile.TemporaryDirectory(
# Cleanup can fail on Windows due to file locks. Ignore if so,
# rather than failing the whole process.
ignore_cleanup_errors=True,
) as temp_dir:
os.environ[ENV_REQUEST_CACHE_PATH] = (
temp_dir # set this as default directory for request_cache to store *.sqlite files
)
Expand Down Expand Up @@ -246,12 +250,18 @@ def handle_record_counts(
) -> AirbyteMessage:
match message.type:
case Type.RECORD:
if message.record is None:
raise ValueError("Record message must have a record attribute")

stream_message_count[
HashableStreamDescriptor(
name=message.record.stream, namespace=message.record.namespace
)
] += 1.0 # type: ignore[union-attr] # record has `stream` and `namespace`
case Type.STATE:
if message.state is None:
raise ValueError("State message must have a state attribute")

stream_descriptor = message_utils.get_stream_descriptor(message)

# Set record count from the counter onto the state message
Expand Down
6 changes: 5 additions & 1 deletion unit_tests/sources/streams/test_call_rate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,11 @@ class StubDummyCacheHttpStream(StubDummyHttpStream):
@pytest.fixture(name="enable_cache")
def enable_cache_fixture():
prev_cache_path = os.environ.get(ENV_REQUEST_CACHE_PATH)
with tempfile.TemporaryDirectory() as temp_dir:
with tempfile.TemporaryDirectory(
# Cleanup can fail on Windows due to file locks. Ignore if so,
# rather than failing the whole process.
ignore_cleanup_errors=True,
) as temp_dir:
os.environ[ENV_REQUEST_CACHE_PATH] = temp_dir
yield

Expand Down
Loading