diff --git a/.gitignore b/.gitignore
index 7c20e4e1..21cc57bb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,3 +14,7 @@ __pycache__/
.venv/
build/
dist/
+
+# The version each package's setup.py reads, written by the publish and by
+# ci/write-package-versions.ps1.
+version.txt
diff --git a/AGENTS.md b/AGENTS.md
index 539b95af..81f3a40b 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -285,6 +285,15 @@ installed the pipeline packages from PyPI. In CI the root
`ci/setup-environment.ps1` installs them from `pipeline/` over those,
so a pipeline change is tested with every product in the same run.
+A package's `setup.py` reads its version from a `version.txt`, which only
+`pipeline/fiftyone_pipeline_did` commits, so any other package installed from
+a checkout is 0.0.0. That is below the minimum versions of the pipeline
+packages that the on-premise and examples packages require, so
+`ci/write-package-versions.ps1` gives each published package the version this
+commit would release it at. The root `ci/setup-environment.ps1` runs it
+first. Run `pwsh ci/write-package-versions.ps1` before running those
+packages' tests from a checkout.
+
## Submodules
The C/C++ engines are the `cxx` submodule, listed in the root `.gitmodules`.
diff --git a/ci/setup-environment.ps1 b/ci/setup-environment.ps1
index 9994e2ef..932f5d81 100644
--- a/ci/setup-environment.ps1
+++ b/ci/setup-environment.ps1
@@ -12,6 +12,10 @@ if ($env:GITHUB_JOB -eq "PreBuild") {
exit 0
}
+# Before anything installs a package from this checkout, so that each one
+# installs at the version this commit would release it at.
+& (Join-Path $PSScriptRoot 'write-package-versions.ps1') -RepoName $RepoName
+
Invoke-ProductStep 'setup-environment' $RepoName $PSBoundParameters
# The device detection, IP intelligence and location scripts install the
diff --git a/ci/write-package-versions.ps1 b/ci/write-package-versions.ps1
new file mode 100644
index 00000000..44b23f25
--- /dev/null
+++ b/ci/write-package-versions.ps1
@@ -0,0 +1,28 @@
+param (
+ [string]$RepoName = '.'
+)
+$ErrorActionPreference = "Stop"
+. (Join-Path $PSScriptRoot 'packages.ps1')
+
+# Writes the version.txt that each published package's setup.py reads. Without
+# one a package installed from this checkout is 0.0.0, which is below the
+# versions the packages here require of each other, so pip refuses it, or
+# replaces it with a release from PyPI. Each package gets the version this
+# commit would release it at, as the publish gives it when it builds the
+# package, or its last release when this commit would not release it.
+$plan = @(Get-ReleasePlan $RepoName)
+$published = @(Get-Packages $RepoName | Where-Object { $_.Publish -eq $true })
+foreach ($package in $published) {
+ # A version.txt committed with the package is left as it is.
+ if (git -C $RepoName ls-files -- "$($package.Path)/version.txt") {
+ continue
+ }
+ $release = $plan | Where-Object { $_.Name -eq $package.Name }
+ if ($null -ne $release) {
+ $version = $release.Version
+ } else {
+ $version = "$((Get-ReleasedVersions $RepoName $package.Name)[-1])"
+ }
+ $version | Out-File (Join-Path $RepoName $package.Path 'version.txt')
+ Write-Output "$($package.Name) $version"
+}
diff --git a/device-detection/ci/run-integration-tests.ps1 b/device-detection/ci/run-integration-tests.ps1
index 9aa2cf2f..f1483528 100644
--- a/device-detection/ci/run-integration-tests.ps1
+++ b/device-detection/ci/run-integration-tests.ps1
@@ -67,7 +67,12 @@ Push-Location $examplesDir
try {
python3 -m venv .virtualenv
if ($LASTEXITCODE -ne 0) { throw "failed to create the examples virtual environment" }
- & $py -m pip install -e .
+ # The pipeline packages come from this checkout rather than a package
+ # index, so the examples run with the pipeline code they are released with.
+ $pipeline = 'fiftyone_pipeline_core', 'fiftyone_pipeline_engines',
+ 'fiftyone_pipeline_engines_fiftyone', 'fiftyone_pipeline_cloudrequestengine'
+ $local = $pipeline | ForEach-Object { '-e'; "../../pipeline/$_" }
+ & $py -m pip install -e . @local
if ($LASTEXITCODE -ne 0) { throw "failed to install the examples package" }
} finally { Pop-Location }
diff --git a/device-detection/fiftyone_devicedetection/examples/gettingstarted.py b/device-detection/fiftyone_devicedetection/examples/gettingstarted.py
index cf3a7b7f..b439c3d3 100644
--- a/device-detection/fiftyone_devicedetection/examples/gettingstarted.py
+++ b/device-detection/fiftyone_devicedetection/examples/gettingstarted.py
@@ -35,46 +35,49 @@
# Cloud
#pipeline = DeviceDetectionPipelineBuilder({"resource_key": resource_key}).build()
-# We create a FlowData object from the pipeline
-# this is used to add evidence to and then process
+# The pipeline is closed at the end of the with block, which releases the
+# native data it holds.
+with pipeline:
+ # We create a FlowData object from the pipeline
+ # this is used to add evidence to and then process. It is closed at the
+ # end of its with block, which releases the results of the detection.
+ with pipeline.create_flowdata() as flowdata1:
-flowdata1 = pipeline.create_flowdata()
+ # Here we add a User-Agent of a desktop as evidence
-# Here we add a User-Agent of a desktop as evidence
+ desktop_ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.97 Safari/537.36"
-desktop_ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.97 Safari/537.36"
+ flowdata1.evidence.add("header.user-agent", desktop_ua)
-flowdata1.evidence.add("header.user-agent", desktop_ua)
+ # Now we process the FlowData
-# Now we process the FlowData
+ flowdata1.process()
-flowdata1.process()
+ # To check whether the User-Agent is a mobile device we look at the ismobile property
+ # inside the Device Detection Engine
-# To check whether the User-Agent is a mobile device we look at the ismobile property
-# inside the Device Detection Engine
+ # first we check if this has a meaningful result
-# first we check if this has a meaningful result
+ print("Is User-Agent " + desktop_ua + " a mobile? ")
+ if flowdata1.device.ismobile.has_value():
+ print(flowdata1.device.ismobile.value())
+ else:
+ # Output why the value isn't meaningful
+ print(flowdata1.device.ismobile.no_value_message())
-print("Is User-Agent " + desktop_ua + " a mobile? ")
-if flowdata1.device.ismobile.has_value():
- print(flowdata1.device.ismobile.value())
-else:
- # Output why the value isn't meaningful
- print(flowdata1.device.ismobile.no_value_message())
+ # Now we do the same with a new User-Agent, this time from an iphone
-# Now we do the same with a new User-Agent, this time from an iphone
+ with pipeline.create_flowdata() as flowdata2:
-flowdata2 = pipeline.create_flowdata()
+ iphone_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"
-iphone_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"
+ flowdata2.evidence.add("header.user-agent", iphone_ua)
-flowdata2.evidence.add("header.user-agent", iphone_ua)
+ flowdata2.process()
-flowdata2.process()
-
-print("Is User-Agent " + iphone_ua + " a mobile device?: ")
-if flowdata2.device.ismobile.has_value():
- print(flowdata2.device.ismobile.value())
-else:
- # Output why the value isn't meaningful
- print(flowdata2.device.ismobile.no_value_message())
+ print("Is User-Agent " + iphone_ua + " a mobile device?: ")
+ if flowdata2.device.ismobile.has_value():
+ print(flowdata2.device.ismobile.value())
+ else:
+ # Output why the value isn't meaningful
+ print(flowdata2.device.ismobile.no_value_message())
diff --git a/device-detection/fiftyone_devicedetection_examples/performance-tests/test_performance.py b/device-detection/fiftyone_devicedetection_examples/performance-tests/test_performance.py
index b7064b14..a8de1b36 100644
--- a/device-detection/fiftyone_devicedetection_examples/performance-tests/test_performance.py
+++ b/device-detection/fiftyone_devicedetection_examples/performance-tests/test_performance.py
@@ -48,16 +48,18 @@ def detect(app, pipeline, user_agents):
# in flowdata.process(). This mock-request doesn't set it, so we set it manually
request.remote_addr = "127.0.0.1"
- flowdata = pipeline.create_flowdata()
+ # The FlowData is closed when the function returns, which releases
+ # the results of the detection straight away.
+ with pipeline.create_flowdata() as flowdata:
- # Add any information from the request (headers, cookies and additional
- # client side provided information)
- flowdata.evidence.add_from_dict(webevidence(request))
+ # Add any information from the request (headers, cookies and additional
+ # client side provided information)
+ flowdata.evidence.add_from_dict(webevidence(request))
- # Process the flowdata
- flowdata.process()
+ # Process the flowdata
+ flowdata.process()
- return getValueHelper(flowdata, "device", "ismobile")
+ return getValueHelper(flowdata, "device", "ismobile")
def benchmark(data_file, user_agents_file, output=False):
app = Flask(__name__)
@@ -73,13 +75,16 @@ def benchmark(data_file, user_agents_file, output=False):
settings = { "add_javascript_builder": False },
).build()
- with open(user_agents_file) as f:
- user_agents_list = [ua.rstrip() for ua in f]
- user_agents = iter(user_agents_list)
+ # The pipeline is closed once the detections have been timed, which
+ # releases the native data it holds.
+ with pipeline:
+ with open(user_agents_file) as f:
+ user_agents_list = [ua.rstrip() for ua in f]
+ user_agents = iter(user_agents_list)
- print(f"Benchmarking with {len(user_agents_list)} user agents", file=sys.stderr)
- environment = {"detect": detect, "app": app, "pipeline": pipeline, "user_agents": user_agents}
- time = timeit("detect(app, pipeline, user_agents)", globals=environment, number=len(user_agents_list))
+ print(f"Benchmarking with {len(user_agents_list)} user agents", file=sys.stderr)
+ environment = {"detect": detect, "app": app, "pipeline": pipeline, "user_agents": user_agents}
+ time = timeit("detect(app, pipeline, user_agents)", globals=environment, number=len(user_agents_list))
if output:
json.dump({
diff --git a/device-detection/fiftyone_devicedetection_examples/setup.py b/device-detection/fiftyone_devicedetection_examples/setup.py
index 15fa344c..e5ab4dbf 100644
--- a/device-detection/fiftyone_devicedetection_examples/setup.py
+++ b/device-detection/fiftyone_devicedetection_examples/setup.py
@@ -65,7 +65,9 @@ def read(file_name):
# named here so that is visible to anyone reading them.
# fiftyone_devicedetection is still required because the on-premise
# examples in this same package use the on-premise engine.
- install_requires=["fiftyone_devicedetection_cloud", "fiftyone_devicedetection", "flask", "flask-unittest", "json5", "ruamel.yaml"],
+ # fiftyone_pipeline_core 4.5.42 is the first version with close(), which
+ # the examples use.
+ install_requires=["fiftyone_devicedetection_cloud", "fiftyone_devicedetection", "fiftyone_pipeline_core>=4.5.42", "flask", "flask-unittest", "json5", "ruamel.yaml"],
license="EUPL-1.2",
classifiers=[
"Development Status :: 5 - Production/Stable",
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py
index d5fd2228..4fd126fc 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py
@@ -234,15 +234,18 @@ def run(self, data_file, license_key, interactive, logger, output):
auto_update = False,
licence_keys = "").add_logger(logger).build()
- # and output the results
- ExampleUtils.check_data_file(pipeline, logger)
- if (ExampleUtils.get_data_file_tier(pipeline.get_element("device")) == "Lite"):
- logger.log("error",
- "Will not download an 'Enterprise' data file over the top of "
- "a 'Lite' data file, please supply another location.")
- raise Exception("File supplied has wrong data tier")
- logger.log("info", "Existing data file will be replaced with downloaded data file")
- logger.log("info", f"Existing data file will be copied to {copy_data_file_name}")
+ # The pipeline is closed at the end of the with block, which
+ # releases the native data it holds before the file is replaced.
+ with pipeline:
+ # and output the results
+ ExampleUtils.check_data_file(pipeline, logger)
+ if (ExampleUtils.get_data_file_tier(pipeline.get_element("device")) == "Lite"):
+ logger.log("error",
+ "Will not download an 'Enterprise' data file over the top of "
+ "a 'Lite' data file, please supply another location.")
+ raise Exception("File supplied has wrong data tier")
+ logger.log("info", "Existing data file will be replaced with downloaded data file")
+ logger.log("info", f"Existing data file will be copied to {copy_data_file_name}")
# do we really want to do this
if (interactive):
@@ -295,41 +298,44 @@ def run(self, data_file, license_key, interactive, logger, output):
data_update_product_type="V4Enterprise",
).add_logger(logger).build()
-
- # thread blocks till update checking is complete - or if there is an
- # exception we don't get this far
- update_event.wait()
- output(f"Update on start-up complete - status - {update_event.status}")
-
- if update_event.status == UpdateStatus.AUTO_UPDATE_SUCCESS:
-
- output("Modifying downloaded file to trigger reload - please wait for that"
- " to complete")
-
- # wait for the dataUpdateService to notify us that it has updated
- update_event.clear()
-
- # it's the same file but changing the file metadata will trigger reload,
- # demonstrating that if you download a new file and replace the
- # existing one, then it will be loaded
- now = datetime.now().timestamp()
- try:
- os.utime(data_file, (now, now))
- except:
- raise Exception("Could not modify file time, abandoning "
- "example")
-
- if update_event.wait(120):
- output(f"Update on file modification complete, status: {update_event.status}")
- else:
- output("Update on file modification timed out")
- else:
- logger.log("error", "Auto update was not successful, abandoning example")
- error_message = f"Auto update failed: {update_event.status}"
- if update_event.status == UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS:
- output(error_message)
+ # The pipeline is closed at the end of the with block, which stops
+ # the automatic updates and the file system watcher, and releases
+ # the native data it holds.
+ with pipeline:
+ # thread blocks till update checking is complete - or if there is an
+ # exception we don't get this far
+ update_event.wait()
+ output(f"Update on start-up complete - status - {update_event.status}")
+
+ if update_event.status == UpdateStatus.AUTO_UPDATE_SUCCESS:
+
+ output("Modifying downloaded file to trigger reload - please wait for that"
+ " to complete")
+
+ # wait for the dataUpdateService to notify us that it has updated
+ update_event.clear()
+
+ # it's the same file but changing the file metadata will trigger reload,
+ # demonstrating that if you download a new file and replace the
+ # existing one, then it will be loaded
+ now = datetime.now().timestamp()
+ try:
+ os.utime(data_file, (now, now))
+ except:
+ raise Exception("Could not modify file time, abandoning "
+ "example")
+
+ if update_event.wait(120):
+ output(f"Update on file modification complete, status: {update_event.status}")
+ else:
+ output("Update on file modification timed out")
else:
- raise Exception(error_message)
+ logger.log("error", "Auto update was not successful, abandoning example")
+ error_message = f"Auto update failed: {update_event.status}"
+ if update_event.status == UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS:
+ output(error_message)
+ else:
+ raise Exception(error_message)
output("Finished Example")
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/failuretomatch.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/failuretomatch.py
index f640b1d8..9db313bd 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/failuretomatch.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/failuretomatch.py
@@ -33,46 +33,49 @@
performance_profile = 'MaxPerformance',
auto_update=False).build()
-# We create a FlowData object from the pipeline
-# this is used to add evidence to and then process
+# The pipeline is closed at the end of the with block, which releases the
+# native data it holds.
+with pipeline:
+ # We create a FlowData object from the pipeline
+ # this is used to add evidence to and then process. It is closed at the
+ # end of its with block, which releases the results of the detection.
+ with pipeline.create_flowdata() as flowdata1:
-flowdata1 = pipeline.create_flowdata()
+ # Here we add a User-Agent of an iphone as evidence
-# Here we add a User-Agent of an iphone as evidence
+ iphone_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"
-iphone_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"
+ flowdata1.evidence.add("header.user-agent", iphone_ua)
-flowdata1.evidence.add("header.user-agent", iphone_ua)
+ # Now we process the FlowData
-# Now we process the FlowData
+ flowdata1.process()
-flowdata1.process()
+ # To check whether the User-Agent is a mobile device we look at the ismobile property
+ # inside the Device Detection Engine
-# To check whether the User-Agent is a mobile device we look at the ismobile property
-# inside the Device Detection Engine
+ # first we check if this has a meaningful result
-# first we check if this has a meaningful result
+ print("Is User-Agent " + iphone_ua + " a mobile device?: ")
+ if flowdata1.device.ismobile.has_value():
+ print(flowdata1.device.ismobile.value())
+ else:
+ # Output why the value isn't meaningful
+ print(flowdata1.device.ismobile.no_value_message())
-print("Is User-Agent " + iphone_ua + " a mobile device?: ")
-if flowdata1.device.ismobile.has_value():
- print(flowdata1.device.ismobile.value())
-else:
- # Output why the value isn't meaningful
- print(flowdata1.device.ismobile.no_value_message())
+ # Now we do the same with a new User-Agent, this time a corrupted one
-# Now we do the same with a new User-Agent, this time a corrupted one
+ badUA = "--"
-badUA = "--"
+ with pipeline.create_flowdata() as flowdata2:
-flowdata2 = pipeline.create_flowdata()
+ flowdata2.evidence.add("header.user-agent", badUA)
-flowdata2.evidence.add("header.user-agent", badUA)
+ flowdata2.process()
-flowdata2.process()
-
-print("Is User-Agent " + badUA + " a mobile device?: ")
-if flowdata2.device.ismobile.has_value():
- print(flowdata2.device.ismobile.value())
-else:
- # Output why the value isn't meaningful
- print(flowdata2.device.ismobile.no_value_message())
+ print("Is User-Agent " + badUA + " a mobile device?: ")
+ if flowdata2.device.ismobile.has_value():
+ print(flowdata2.device.ismobile.value())
+ else:
+ # Output why the value isn't meaningful
+ print(flowdata2.device.ismobile.no_value_message())
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_console.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_console.py
index 946f06bd..3675fee2 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_console.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_console.py
@@ -68,11 +68,16 @@ def run(self, data_file, logger, output):
auto_update = False,
licence_keys = "").add_logger(logger).build()
- ExampleUtils.check_data_file(pipeline, logger)
+ # The pipeline is closed at the end of the with block, which
+ # releases the native data it holds. An application normally builds
+ # one pipeline when it starts, uses it for every detection, and
+ # closes it when it stops.
+ with pipeline:
+ ExampleUtils.check_data_file(pipeline, logger)
- # carry out some sample detections
- for values in EVIDENCE_VALUES:
- self.analyseEvidence(values, pipeline, output)
+ # carry out some sample detections
+ for values in EVIDENCE_VALUES:
+ self.analyseEvidence(values, pipeline, output)
def analyseEvidence(self, evidence, pipeline, output):
@@ -83,42 +88,45 @@ def analyseEvidence(self, evidence, pipeline, output):
# and usually consists of a number of HTTP Header field
# values, in this case represented by a dictionary of header
# name/value entries.
- data = pipeline.create_flowdata()
+ # Closing the FlowData, which the with statement does at the end of
+ # the block, releases the results of the detection straight away,
+ # rather than when the garbage collector next runs.
+ with pipeline.create_flowdata() as data:
- message = []
+ message = []
- # List the evidence
- message.append("Input values:\n")
- for key in evidence:
- message.append(f"\t{key}: {evidence[key]}\n")
+ # List the evidence
+ message.append("Input values:\n")
+ for key in evidence:
+ message.append(f"\t{key}: {evidence[key]}\n")
- output("".join(message))
-
- # Add the evidence values to the flow data
- data.evidence.add_from_dict(evidence)
-
- # Process the flow data.
- data.process()
-
- message = []
- message.append("Results:\n")
-
- # Now that it's been processed, the flow data will have
- # been populated with the result. In this case, we want
- # information about the device, which we can get by
- # asking for a result matching named "device"
- device = data.device
-
- # Display the results of the detection, which are called
- # device properties. See the property dictionary at
- # https://51degrees.com/developers/property-dictionary?utm_source=code&utm_medium=example&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_examples-src-fiftyone_devicedetection_examples-onpremise-gettingstarted_console.py&utm_term=analyseevidence
- # for details of all available properties.
- self.outputValue("Mobile Device", device.ismobile, message)
- self.outputValue("Platform Name", device.platformname, message)
- self.outputValue("Platform Version", device.platformversion, message)
- self.outputValue("Browser Name", device.browsername, message)
- self.outputValue("Browser Version", device.browserversion, message)
- output("".join(message))
+ output("".join(message))
+
+ # Add the evidence values to the flow data
+ data.evidence.add_from_dict(evidence)
+
+ # Process the flow data.
+ data.process()
+
+ message = []
+ message.append("Results:\n")
+
+ # Now that it's been processed, the flow data will have
+ # been populated with the result. In this case, we want
+ # information about the device, which we can get by
+ # asking for a result matching named "device"
+ device = data.device
+
+ # Display the results of the detection, which are called
+ # device properties. See the property dictionary at
+ # https://51degrees.com/developers/property-dictionary?utm_source=code&utm_medium=example&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_examples-src-fiftyone_devicedetection_examples-onpremise-gettingstarted_console.py&utm_term=analyseevidence
+ # for details of all available properties.
+ self.outputValue("Mobile Device", device.ismobile, message)
+ self.outputValue("Platform Name", device.platformname, message)
+ self.outputValue("Platform Version", device.platformversion, message)
+ self.outputValue("Browser Name", device.browsername, message)
+ self.outputValue("Browser Version", device.browserversion, message)
+ output("".join(message))
def outputValue(self, name, value, message):
# Individual result values have a wrapper called
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_web/app.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_web/app.py
index d382e6f2..a8e1e3ce 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_web/app.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/gettingstarted_web/app.py
@@ -102,7 +102,10 @@ def run(self):
# The port can be set with the PORT environment variable, as it can
# for the cloud example, so the example can run next to other services.
- GettingStartedWeb.app.run(port=int(os.environ.get("PORT", 5001)))
+ # The pipeline lives as long as the app, and is closed when the server
+ # stops, which releases the native data it holds.
+ with GettingStartedWeb.pipeline:
+ GettingStartedWeb.app.run(port=int(os.environ.get("PORT", 5001)))
# First we make a JSON route that will be called from the client side and will return
# a JSON encoded property database using any additional evidence provided by the client
@@ -111,21 +114,23 @@ def run(self):
@app.route('/json', methods=['POST'])
def jsonroute():
- # Create the flowdata object for the JSON route
- flowdata = GettingStartedWeb.pipeline.create_flowdata()
+ # Create the flowdata object for the JSON route. It is closed when the
+ # route returns, which releases the results of the detection straight
+ # away rather than when the garbage collector next runs.
+ with GettingStartedWeb.pipeline.create_flowdata() as flowdata:
- # Add any information from the request (headers, cookies and additional
- # client side provided information)
+ # Add any information from the request (headers, cookies and additional
+ # client side provided information)
- flowdata.evidence.add_from_dict(webevidence(request))
+ flowdata.evidence.add_from_dict(webevidence(request))
- # Process the flowdata
+ # Process the flowdata
- flowdata.process()
+ flowdata.process()
- # Return the JSON from the JSONBundler engine
+ # Return the JSON from the JSONBundler engine
- return json.dumps(flowdata.jsonbundler.json)
+ return json.dumps(flowdata.jsonbundler.json)
# Next we serve the client-side JavaScript from its own route. The other
# Pipeline APIs ship a web integration that intercepts '/51Degrees.core.js'
@@ -136,26 +141,27 @@ def jsonroute():
@app.route('/51Degrees.core.js')
def core_js():
- # Create the flowdata object for the JavaScript route
- flowdata = GettingStartedWeb.pipeline.create_flowdata()
+ # Create the flowdata object for the JavaScript route. It is closed
+ # when the route returns.
+ with GettingStartedWeb.pipeline.create_flowdata() as flowdata:
- # Add any information from the request (headers, cookies and additional
- # client side provided information). Query parameters are included, so
- # the per-request 'fod-js-enable-cookies' override is picked up here and
- # applied by the JavaScriptBuilder engine.
+ # Add any information from the request (headers, cookies and additional
+ # client side provided information). Query parameters are included, so
+ # the per-request 'fod-js-enable-cookies' override is picked up here and
+ # applied by the JavaScriptBuilder engine.
- flowdata.evidence.add_from_dict(webevidence(request))
+ flowdata.evidence.add_from_dict(webevidence(request))
- # Process the flowdata
+ # Process the flowdata
- flowdata.process()
+ flowdata.process()
- # Return the JavaScript from the JavaScriptBuilder engine
+ # Return the JavaScript from the JavaScriptBuilder engine
- response = make_response(flowdata.javascriptbuilder.javascript)
- response.headers["Content-Type"] = "application/x-javascript"
+ response = make_response(flowdata.javascriptbuilder.javascript)
+ response.headers["Content-Type"] = "application/x-javascript"
- return response
+ return response
# In the main route we dynamically update the screen's device property display
# using the above JSON route
@@ -164,36 +170,37 @@ def core_js():
@app.route('/')
def server():
- # Create the flowdata object for the JSON route
- flowdata = GettingStartedWeb.pipeline.create_flowdata()
+ # Create the flowdata object for the page. It is closed when the route
+ # returns, once the page has been rendered from it.
+ with GettingStartedWeb.pipeline.create_flowdata() as flowdata:
- # Add any information from the request (headers, cookies and additional
- # client side provided information)
+ # Add any information from the request (headers, cookies and additional
+ # client side provided information)
- flowdata.evidence.add_from_dict(webevidence(request))
+ flowdata.evidence.add_from_dict(webevidence(request))
- # Process the flowdata
+ # Process the flowdata
- flowdata.process()
+ flowdata.process()
- response = make_response()
+ response = make_response()
- # Some browsers require that extra HTTP headers are explicitly
- # requested. So set whatever headers are required by the browser in
- # order to return the evidence needed by the pipeline.
- # More info on this can be found at
- # https://51degrees.com/blog/user-agent-client-hints?utm_source=code&utm_medium=example&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_examples-src-fiftyone_devicedetection_examples-onpremise-gettingstarted_web-app.py&utm_term=server
+ # Some browsers require that extra HTTP headers are explicitly
+ # requested. So set whatever headers are required by the browser in
+ # order to return the evidence needed by the pipeline.
+ # More info on this can be found at
+ # https://51degrees.com/blog/user-agent-client-hints?utm_source=code&utm_medium=example&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_examples-src-fiftyone_devicedetection_examples-onpremise-gettingstarted_web-app.py&utm_term=server
- set_response_header(flowdata, response)
+ set_response_header(flowdata, response)
- # Generate the HTML
- response.set_data(render_template(
- 'index.html',
- data=flowdata,
- utils=ExampleUtils,
- response=response))
+ # Generate the HTML
+ response.set_data(render_template(
+ 'index.html',
+ data=flowdata,
+ utils=ExampleUtils,
+ response=response))
- return response
+ return response
# Typically, something like this will not be necessary.
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/match_metrics.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/match_metrics.py
index 5daeb284..6a223d00 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/match_metrics.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/match_metrics.py
@@ -88,83 +88,89 @@ def run(self, data_file, show_descs, logger, output):
# We want to show the matching evidence characters as part of this example, so we have to set
# this flag to true.
update_matched_useragent = True).add_logger(logger).build()
+
+ # The pipeline is closed at the end of the with block, which
+ # releases the native data it holds.
+ with pipeline:
+ ExampleUtils.check_data_file(pipeline, logger)
+
+ # Closing the FlowData, which the with statement does at the end
+ # of the block, releases the results of the detection straight
+ # away, rather than when the garbage collector next runs.
+ with pipeline.create_flowdata() as data:
+
+ # Process a single evidence to retrieve the values
+ # associated with the user-agent and other evidence such as sec-ch-* for the
+ # selected properties.
+ data.evidence.add_from_dict(self.Evidence)
+ data.process()
+
+ device = data.device
+
+ output("--- Compare evidence with what was matched ---\n")
+ output("Evidence")
+ # output the evidence in reverse value length order
+ for entry in sorted(self.Evidence.items(), key=lambda item: len(item[1]), reverse=True):
+ output(f" {entry[0]}: {entry[1]}")
+ # Obtain the matched User-Agents: the matched substrings in the
+ # User-Agents are separated with underscores - output in forward length order.
+ output("Matches")
+ for entry in sorted(device.useragents.value(), key=lambda item: len(item)):
+ output(f" Matched User-Agent: {entry}")
+
+ output("")
+
+
+ output("--- Listing all available properties, by component, by property name ---")
+ output("For a discussion of what the match properties mean, see: https://51degrees.com/documentation/_device_detection__hash.html?utm_source=code&utm_medium=example&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_examples-src-fiftyone_devicedetection_examples-onpremise-match_metrics.py&utm_term=match-properties-discussion#DeviceDetection_Hash_DataSetProduction_Performance\n")
+
+ # get the properties available from the DeviceDetection engine
+ # which has the key "device". For the sake of illustration we will
+ # retrieve it indirectly.
+ hashEngineElementKey = pipeline.get_element("device").datakey
+
+ # retrieve the available properties from the hash engine. The properties
+ # available depends on
+ # a) the use of restricted_properties in the builder (see above)
+ # which controls which properties will be extracted, and also affects
+ # the performance of extraction
+ # b) the tier of data file being used. The Lite data file contains fewer
+ # than 20 of the >200 available properties
+ availableProperties = dict(pipeline.get_properties()[hashEngineElementKey])
- ExampleUtils.check_data_file(pipeline, logger)
-
- data = pipeline.create_flowdata()
-
- # Process a single evidence to retrieve the values
- # associated with the user-agent and other evidence such as sec-ch-* for the
- # selected properties.
- data.evidence.add_from_dict(self.Evidence)
- data.process()
-
- device = data.device
-
- output("--- Compare evidence with what was matched ---\n")
- output("Evidence")
- # output the evidence in reverse value length order
- for entry in sorted(self.Evidence.items(), key=lambda item: len(item[1]), reverse=True):
- output(f" {entry[0]}: {entry[1]}")
- # Obtain the matched User-Agents: the matched substrings in the
- # User-Agents are separated with underscores - output in forward length order.
- output("Matches")
- for entry in sorted(device.useragents.value(), key=lambda item: len(item)):
- output(f" Matched User-Agent: {entry}")
-
- output("")
-
-
- output("--- Listing all available properties, by component, by property name ---")
- output("For a discussion of what the match properties mean, see: https://51degrees.com/documentation/_device_detection__hash.html?utm_source=code&utm_medium=example&utm_campaign=51degrees-python&utm_content=fiftyone_devicedetection_examples-src-fiftyone_devicedetection_examples-onpremise-match_metrics.py&utm_term=match-properties-discussion#DeviceDetection_Hash_DataSetProduction_Performance\n")
-
- # get the properties available from the DeviceDetection engine
- # which has the key "device". For the sake of illustration we will
- # retrieve it indirectly.
- hashEngineElementKey = pipeline.get_element("device").datakey
-
- # retrieve the available properties from the hash engine. The properties
- # available depends on
- # a) the use of restricted_properties in the builder (see above)
- # which controls which properties will be extracted, and also affects
- # the performance of extraction
- # b) the tier of data file being used. The Lite data file contains fewer
- # than 20 of the >200 available properties
- availableProperties = dict(pipeline.get_properties()[hashEngineElementKey])
-
- # create a Map keyed on the component name of the properties available
- # components being hardware, browser, OS and Crawler.
- def get_component(property):
- return property[1]["component"]
-
- availableProperties = dict(sorted(availableProperties.items(),key=lambda item: get_component(item)))
- categoryMap = groupby(availableProperties.items(), get_component)
- # iterate the map created above
- for component, properties in categoryMap:
- output(component)
- for propertyKey, property in properties:
- propertyName = property["name"]
- propertyDescription = property["description"]
+ # create a Map keyed on the component name of the properties available
+ # components being hardware, browser, OS and Crawler.
+ def get_component(property):
+ return property[1]["component"]
+
+ availableProperties = dict(sorted(availableProperties.items(),key=lambda item: get_component(item)))
+ categoryMap = groupby(availableProperties.items(), get_component)
+ # iterate the map created above
+ for component, properties in categoryMap:
+ output(component)
+ for propertyKey, property in properties:
+ propertyName = property["name"]
+ propertyDescription = property["description"]
- # while we get the available properties and their metadata from the
- # pipeline we get the values for the last detection from flowData
- value = device[propertyKey]
+ # while we get the available properties and their metadata from the
+ # pipeline we get the values for the last detection from flowData
+ value = device[propertyKey]
- # output property names, values and descriptions
- # some property values are lists.
- if value.has_value() and isinstance(value.value(), list):
- output(f" {propertyName}: {len(value.value())} Values")
- for item in value.value():
- output(f" : {item}")
+ # output property names, values and descriptions
+ # some property values are lists.
+ if value.has_value() and isinstance(value.value(), list):
+ output(f" {propertyName}: {len(value.value())} Values")
+ for item in value.value():
+ output(f" : {item}")
- else:
- output(f" {propertyName}: {value.value()}")
+ else:
+ output(f" {propertyName}: {value.value()}")
- if (show_descs == True):
- output(f" {propertyDescription}")
+ if (show_descs == True):
+ output(f" {propertyDescription}")
- output()
- logger.log("info", "Finished Match Metrics Example")
+ output()
+ logger.log("info", "Finished Match Metrics Example")
# Evidence values from a windows 11 device using a browser
# that supports User-Agent Client Hints.
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/metadata_console.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/metadata_console.py
index 2df9d104..5e605158 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/metadata_console.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/metadata_console.py
@@ -76,10 +76,14 @@ def run(self, data_file, logger, output):
# Inhibit auto-update of the data file for this example
auto_update = False,
licence_keys = "").add_logger(logger).build()
- self.outputProperties(pipeline.get_element("device"), output)
- self.outputEvidenceKeyDetails(pipeline.get_element("device"), output)
- ExampleUtils.check_data_file(pipeline, logger)
+ # The pipeline is closed at the end of the with block, which
+ # releases the native data it holds.
+ with pipeline:
+ self.outputProperties(pipeline.get_element("device"), output)
+ self.outputEvidenceKeyDetails(pipeline.get_element("device"), output)
+
+ ExampleUtils.check_data_file(pipeline, logger)
def outputEvidenceKeyDetails(self, engine, output):
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/offlineprocessing.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/offlineprocessing.py
index 7a90c401..500fdba7 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/offlineprocessing.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/offlineprocessing.py
@@ -99,40 +99,43 @@ def run(self, data_file, evidence_yaml, logger, output):
auto_update = False,
licence_keys = "").add_logger(logger).build()
- records = 0
- yaml = YAML()
- # Replace tab characters as they are not valid in unquoted YAML values
- # and some evidence files contain them in user-agent strings.
- content = evidence_yaml.read().replace('\t', ' ')
- yaml_data = yaml.load_all(content)
+ # The pipeline is closed at the end of the with block, which
+ # releases the native data it holds.
+ with pipeline:
+ records = 0
+ yaml = YAML()
+ # Replace tab characters as they are not valid in unquoted YAML values
+ # and some evidence files contain them in user-agent strings.
+ content = evidence_yaml.read().replace('\t', ' ')
+ yaml_data = yaml.load_all(content)
- try:
- # Keep going as long as we have more document records.
- for evidence in yaml_data:
- # Output progress.
- records = records + 1
- if (records % 100 == 0):
- logger.log("info", f"Processed {records} records")
+ try:
+ # Keep going as long as we have more document records.
+ for evidence in yaml_data:
+ # Output progress.
+ records = records + 1
+ if (records % 100 == 0):
+ logger.log("info", f"Processed {records} records")
- # write the yaml document separator
- print("---", file = output)
- # Pass the record to the pipeline as evidence so that it can be analyzed
- headers = {}
- for key in evidence:
- # Convert to string - YAML may parse values like ?1 as booleans
- # Keys already have header. prefix in the YAML file
- headers[key] = str(evidence[key])
+ # write the yaml document separator
+ print("---", file = output)
+ # Pass the record to the pipeline as evidence so that it can be analyzed
+ headers = {}
+ for key in evidence:
+ # Convert to string - YAML may parse values like ?1 as booleans
+ # Keys already have header. prefix in the YAML file
+ headers[key] = str(evidence[key])
- self.analyseEvidence(headers, pipeline, output, yaml)
- except BaseException as err:
- # We can't read the evidence values, so cant write them to the output. Will just
- # have to skip this entry.
- logger.log("error", err)
+ self.analyseEvidence(headers, pipeline, output, yaml)
+ except BaseException as err:
+ # We can't read the evidence values, so cant write them to the output. Will just
+ # have to skip this entry.
+ logger.log("error", err)
- # write the yaml document end marker
- print("...", file = output)
+ # write the yaml document end marker
+ print("...", file = output)
- ExampleUtils.check_data_file(pipeline, logger)
+ ExampleUtils.check_data_file(pipeline, logger)
def analyseEvidence(self, evidence, pipeline, output, yaml):
# FlowData is a data structure that is used to convey information required for
@@ -140,46 +143,50 @@ def analyseEvidence(self, evidence, pipeline, output, yaml):
# Information required for detection is called "evidence" and usually consists
# of a number of HTTP Header field values, in this case represented by a
# dictionary of header name/value entries.
- data = pipeline.create_flowdata()
- # Add the evidence values to the flow data
- data.evidence.add_from_dict(evidence)
- # Process the flow data.
- data.process()
+ # Closing the FlowData, which the with statement does at the end of
+ # the block, releases the results of the detection straight away.
+ # Without it each record's results would stay in memory until the
+ # garbage collector next runs.
+ with pipeline.create_flowdata() as data:
+ # Add the evidence values to the flow data
+ data.evidence.add_from_dict(evidence)
+ # Process the flow data.
+ data.process()
- device = data.device
+ device = data.device
- values = {}
- # Add the evidence values to the output
- for key in evidence:
- values[key] = evidence[key]
- # Now add the values that we want to store against the record.
- values["device.ismobile"] = device.ismobile.value() if device.ismobile.has_value() else "Unknown"
- values["device.platformname"] = ExampleUtils.get_human_readable(device, "platformname")
- values["device.platformversion"] = ExampleUtils.get_human_readable(device, "platformversion")
- values["device.browsername"] = ExampleUtils.get_human_readable(device, "browsername")
- values["device.browserversion"] = ExampleUtils.get_human_readable(device, "browserversion")
- values["device.tac"] = ExampleUtils.get_human_readable(device, "tac")
- values["device.devicetype"] = ExampleUtils.get_human_readable(device, "devicetype")
- values["device.hardwarevendor"] = ExampleUtils.get_human_readable(device, "hardwarevendor")
- values["device.hardwarename"] = ExampleUtils.get_human_readable(device, "hardwarename")
- values["device.hardwaremodel"] = ExampleUtils.get_human_readable(device, "hardwaremodel")
- values["device.nativedevice"] = ExampleUtils.get_human_readable(device, "nativedevice")
- values["device.nativename"] = ExampleUtils.get_human_readable(device, "nativename")
- values["device.nativemodel"] = ExampleUtils.get_human_readable(device, "nativemodel")
- # DeviceId is a unique identifier for the combination of hardware, operating
- # system, browser and crawler that has been detected.
- # Our device detection solution uses machine learning to find the optimal
- # way to identify devices based on the real-world evidence values that we
- # observe each day.
- # As this changes over time, the result of detection can potentially change
- # as well. By storing the device id, we can use this as a lookup in future
- # rather than performing detection with the original evidence again.
- # Do this by passing an evidence entry with:
- # key = query.51D_ProfileIds
- # value = [the device id]
- # This is much faster and avoids the potential for getting a different
- # result.
- values["device.deviceid"] = ExampleUtils.get_human_readable(device, "deviceid")
+ values = {}
+ # Add the evidence values to the output
+ for key in evidence:
+ values[key] = evidence[key]
+ # Now add the values that we want to store against the record.
+ values["device.ismobile"] = device.ismobile.value() if device.ismobile.has_value() else "Unknown"
+ values["device.platformname"] = ExampleUtils.get_human_readable(device, "platformname")
+ values["device.platformversion"] = ExampleUtils.get_human_readable(device, "platformversion")
+ values["device.browsername"] = ExampleUtils.get_human_readable(device, "browsername")
+ values["device.browserversion"] = ExampleUtils.get_human_readable(device, "browserversion")
+ values["device.tac"] = ExampleUtils.get_human_readable(device, "tac")
+ values["device.devicetype"] = ExampleUtils.get_human_readable(device, "devicetype")
+ values["device.hardwarevendor"] = ExampleUtils.get_human_readable(device, "hardwarevendor")
+ values["device.hardwarename"] = ExampleUtils.get_human_readable(device, "hardwarename")
+ values["device.hardwaremodel"] = ExampleUtils.get_human_readable(device, "hardwaremodel")
+ values["device.nativedevice"] = ExampleUtils.get_human_readable(device, "nativedevice")
+ values["device.nativename"] = ExampleUtils.get_human_readable(device, "nativename")
+ values["device.nativemodel"] = ExampleUtils.get_human_readable(device, "nativemodel")
+ # DeviceId is a unique identifier for the combination of hardware, operating
+ # system, browser and crawler that has been detected.
+ # Our device detection solution uses machine learning to find the optimal
+ # way to identify devices based on the real-world evidence values that we
+ # observe each day.
+ # As this changes over time, the result of detection can potentially change
+ # as well. By storing the device id, we can use this as a lookup in future
+ # rather than performing detection with the original evidence again.
+ # Do this by passing an evidence entry with:
+ # key = query.51D_ProfileIds
+ # value = [the device id]
+ # This is much faster and avoids the potential for getting a different
+ # result.
+ values["device.deviceid"] = ExampleUtils.get_human_readable(device, "deviceid")
yaml.dump(values, output)
def main(argv):
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/performance.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/performance.py
index 271ce134..d795b548 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/performance.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/performance.py
@@ -62,24 +62,26 @@
def process_user_agent(user_agent):
- # First we create the flowdata using the global pipeline
- flowdata = pipeline.create_flowdata() # pylint: disable=used-before-assignment
+ # First we create the flowdata using the global pipeline. Closing it,
+ # which the with statement does when the function returns, releases the
+ # results of the detection straight away.
+ with pipeline.create_flowdata() as flowdata: # pylint: disable=used-before-assignment
- # Here we add the user agent as evidence
- flowdata.evidence.add("header.user-agent", user_agent)
+ # Here we add the user agent as evidence
+ flowdata.evidence.add("header.user-agent", user_agent)
- # We process the flowdata to get the results
- flowdata.process()
+ # We process the flowdata to get the results
+ flowdata.process()
- # To check whether the User-Agent is a mobile device we look at the ismobile
- # property inside the Device Detection Engine
+ # To check whether the User-Agent is a mobile device we look at the ismobile
+ # property inside the Device Detection Engine
- # first we check if this has a meaningful result
+ # first we check if this has a meaningful result
- if flowdata.device.ismobile.has_value():
- return flowdata.device.ismobile.value()
- else:
- return None
+ if flowdata.device.ismobile.has_value():
+ return flowdata.device.ismobile.value()
+ else:
+ return None
def process_user_agent_list(data_file, user_agent_list, list_number, output, skip=False):
global pipeline
@@ -92,23 +94,26 @@ def process_user_agent_list(data_file, user_agent_list, list_number, output, ski
usage_sharing=False,
auto_update=False).build()
- results = {
- "mobile": 0,
- "notmobile": 0,
- "unknown": 0
- }
- for user_agent in user_agent_list:
- if skip:
- break
- result = process_user_agent(user_agent[0])
- if(result == None):
- results["unknown"] += 1
- if(result == True):
- results["mobile"] += 1
- if(result == False):
- results["notmobile"] += 1
-
- output.put(results, list_number)
+ # The pipeline is closed at the end of the with block, which releases
+ # the native data it holds.
+ with pipeline:
+ results = {
+ "mobile": 0,
+ "notmobile": 0,
+ "unknown": 0
+ }
+ for user_agent in user_agent_list:
+ if skip:
+ break
+ result = process_user_agent(user_agent[0])
+ if(result == None):
+ results["unknown"] += 1
+ if(result == True):
+ results["mobile"] += 1
+ if(result == False):
+ results["notmobile"] += 1
+
+ output.put(results, list_number)
# Run the process
def run(data_file, skip = False):
diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/shareusage.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/shareusage.py
index 6cbbbfd7..51bcff0c 100644
--- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/shareusage.py
+++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/shareusage.py
@@ -130,46 +130,51 @@ def run(self, evidence_yaml):
# Build a pipeline with only the ShareUsage element
pipeline = PipelineBuilder().add(share_usage_element).build()
- print("Pipeline built successfully with ShareUsage element")
-
- # Load YAML evidence
- yaml = YAML()
- yaml_data = yaml.load_all(evidence_yaml)
-
- # Process evidence records
- count = 0
- for evidence in yaml_data:
- if count >= RECORDS_TO_PROCESS:
- break
-
- # Filter to only header.* entries
- data = filter_evidence(
- {k: str(v) for k, v in evidence.items()},
- "header.")
-
- # Add a client IP address to the evidence for deduplication purposes
- # (see class-level note). In a real web application, this would come
- # from the request (e.g., REMOTE_ADDR or X-Forwarded-For header)
- data[EVIDENCE_CLIENTIP_KEY] = generate_random_ip()
-
- # Add the usage-from identifier so 51Degrees knows the source of this data.
- # This appears as
IP address: {client_ip}
-Most probable country first, then every other country alphabetically.
- {country_dropdown(translated)} -Add ?client-ip=x.x.x.x to the url to lookup a different address, and - &translation=fr_FR to choose a language.
- """ + # Create the flowdata object for this request. It is closed when the + # route returns, once the page has been built from it, which releases + # the results of the lookup straight away. + with pipeline.create_flowdata() as flowdata: + + # Use the IP address supplied as a query parameter, or fall back + # to the address of the requester. + client_ip = request.args.get("client-ip", request.remote_addr) + flowdata.evidence.add("query.client-ip", client_ip) + + # Supply the browser language so the country names are localized. The + # Accept-Language header is used by default, overridable with a + # "translation" query parameter. + accept_language = request.headers.get("Accept-Language") + if accept_language: + flowdata.evidence.add("header.accept-language", accept_language) + translation = request.args.get("translation") + if translation: + flowdata.evidence.add("query.translation", translation) + + # Process the flowdata + flowdata.process() + + ip_data = flowdata.ip + translated = flowdata.get("countrynamestranslated") + + return f""" +IP address: {client_ip}
+Most probable country first, then every other country alphabetically.
+ {country_dropdown(translated)} +Add ?client-ip=x.x.x.x to the url to lookup a different address, and + &translation=fr_FR to choose a language.
+ """ def main(data_file = None): @@ -138,8 +140,10 @@ def main(data_file = None): "`git submodule update --init --recursive`, and run this " "example again to download the file, or supply the path to a " "data file as the first command line argument.") - build_pipeline(data_file) - app.run(debug=False) + # The pipeline lives as long as the app, and is closed when the server + # stops, which releases the native data it holds. + with build_pipeline(data_file): + app.run(debug=False) if __name__ == "__main__": diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/metadata_console.py b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/metadata_console.py index ec2be8a1..3819776c 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/metadata_console.py +++ b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/metadata_console.py @@ -47,14 +47,17 @@ def run(self, data_file, logger, output): auto_update = False, licence_keys = "").add_logger(logger).build() - engine = pipeline.flow_elements_list["ip"] + # The pipeline is closed at the end of the with block, which + # releases the native data it holds. + with pipeline: + engine = pipeline.flow_elements_list["ip"] - output("IP Intelligence properties:") - for property_metadata in engine.properties.values(): - output(f" {property_metadata['name']} ({property_metadata['type']}) - " - f"category: {property_metadata['category']}") - if property_metadata.get("description"): - output(f" {property_metadata['description']}") + output("IP Intelligence properties:") + for property_metadata in engine.properties.values(): + output(f" {property_metadata['name']} ({property_metadata['type']}) - " + f"category: {property_metadata['category']}") + if property_metadata.get("description"): + output(f" {property_metadata['description']}") def main(argv): diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/mixed_console.py b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/mixed_console.py index 07839ab1..2f4f5e18 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/mixed_console.py +++ b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/mixed_console.py @@ -140,38 +140,43 @@ def run(self, device_file, ip_file, logger, output): .add_logger(logger) \ .build() - for evidence in COMBINED_EVIDENCE: - self.analyse_evidence(evidence, pipeline, output) + # The pipeline is closed at the end of the with block, which closes + # both engines and releases the native data they hold. + with pipeline: + for evidence in COMBINED_EVIDENCE: + self.analyse_evidence(evidence, pipeline, output) def analyse_evidence(self, evidence, pipeline, output): - data = pipeline.create_flowdata() + # The FlowData is closed at the end of the with block, which releases + # the results of both engines straight away. + with pipeline.create_flowdata() as data: - message = [] - message.append("=" * 70 + "\n") - message.append("Input values:\n") - for key in evidence: - message.append(f"\t{key}: {evidence[key]}\n") - output("".join(message)) + message = [] + message.append("=" * 70 + "\n") + message.append("Input values:\n") + for key in evidence: + message.append(f"\t{key}: {evidence[key]}\n") + output("".join(message)) - # Add the evidence values to the flow data and process it. Both - # engines run against the same flow data. - data.evidence.add_from_dict(evidence) - data.process() + # Add the evidence values to the flow data and process it. Both + # engines run against the same flow data. + data.evidence.add_from_dict(evidence) + data.process() - message = [] + message = [] - # Device detection results, keyed under "device". - message.append("Device Detection Results:\n") - for name in DEVICE_PROPERTIES: - self.outputValue(data.device, name, message) + # Device detection results, keyed under "device". + message.append("Device Detection Results:\n") + for name in DEVICE_PROPERTIES: + self.outputValue(data.device, name, message) - # IP intelligence results, keyed under "ip". - message.append("\nIP Intelligence Results:\n") - for name in IP_PROPERTIES: - self.outputValue(data.ip, name, message) + # IP intelligence results, keyed under "ip". + message.append("\nIP Intelligence Results:\n") + for name in IP_PROPERTIES: + self.outputValue(data.ip, name, message) - output("".join(message)) + output("".join(message)) def outputValue(self, aspect_data, name, message): # `get_value` returns None when the property is not present in the diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/offlineprocessing.py b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/offlineprocessing.py index 9c7109cc..4785249e 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/offlineprocessing.py +++ b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/offlineprocessing.py @@ -65,41 +65,47 @@ def run(self, data_file, evidence_yaml, logger, output): auto_update = False, licence_keys = "").add_logger(logger).build() - ExampleUtils.check_data_file(pipeline, logger) - - records = 0 - with open(evidence_yaml, "r") as fin: - for evidence in yaml.load_all(fin): - if records >= RECORD_LIMIT: - break - - # write the yaml document separator - output("---") - self.analyse_evidence(evidence, pipeline, output, yaml) - records += 1 - - # write the yaml document end marker - output("...") - logger.log("info", f"Processed {records} evidence records") + # The pipeline is closed at the end of the with block, which + # releases the native data it holds. + with pipeline: + ExampleUtils.check_data_file(pipeline, logger) + + records = 0 + with open(evidence_yaml, "r") as fin: + for evidence in yaml.load_all(fin): + if records >= RECORD_LIMIT: + break + + # write the yaml document separator + output("---") + self.analyse_evidence(evidence, pipeline, output, yaml) + records += 1 + + # write the yaml document end marker + output("...") + logger.log("info", f"Processed {records} evidence records") def analyse_evidence(self, evidence, pipeline, output, yaml): - data = pipeline.create_flowdata() + # The FlowData is closed at the end of the with block, once the + # values needed have been read, which releases the results of the + # lookup straight away. + with pipeline.create_flowdata() as data: - # Add the evidence values to the flow data. The YAML loader - # returns a dict subclass so convert to a plain dict first. - data.evidence.add_from_dict(dict(evidence)) + # Add the evidence values to the flow data. The YAML loader + # returns a dict subclass so convert to a plain dict first. + data.evidence.add_from_dict(dict(evidence)) - # Process the flow data. - data.process() + # Process the flow data. + data.process() - ip_data = data.ip + ip_data = data.ip - result = dict(evidence) + result = dict(evidence) - # Add the values that we want to store against the record. - result["ip.registeredname"] = ExampleUtils.get_human_readable(ip_data, "registeredname") - result["ip.registeredcountry"] = ExampleUtils.get_human_readable(ip_data, "registeredcountry") + # Add the values that we want to store against the record. + result["ip.registeredname"] = ExampleUtils.get_human_readable(ip_data, "registeredname") + result["ip.registeredcountry"] = ExampleUtils.get_human_readable(ip_data, "registeredcountry") buffer = io.BytesIO() yaml.dump(result, buffer) diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/performance_console.py b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/performance_console.py index 310bd727..1d910d81 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/performance_console.py +++ b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/performance_console.py @@ -84,11 +84,13 @@ def _build_pipeline(data_file, performance_profile, logger=None): def _process_record(pipeline, record): - data = pipeline.create_flowdata() - data.evidence.add_from_dict(record) - data.process() - value = ExampleUtils.get_value(data.ip, PERFORMANCE_PROPERTY) - return value is not None and value.has_value() + # The FlowData is closed when the function returns, which releases the + # results of the lookup straight away. + with pipeline.create_flowdata() as data: + data.evidence.add_from_dict(record) + data.process() + value = ExampleUtils.get_value(data.ip, PERFORMANCE_PROPERTY) + return value is not None and value.has_value() def _time_lookups(pipeline, records): @@ -115,8 +117,10 @@ def _worker(data_file, performance_profile, records, queue): count and elapsed seconds back to the parent. """ - pipeline = _build_pipeline(data_file, performance_profile) - elapsed = _time_lookups(pipeline, records) + # The pipeline is closed at the end of the with block, which releases + # the native data it holds. + with _build_pipeline(data_file, performance_profile) as pipeline: + elapsed = _time_lookups(pipeline, records) queue.put((len(records), elapsed)) @@ -156,9 +160,10 @@ def run(self, data_file, evidence, logger, output, workers=None, # wall time of the run. elapsed = max(elapsed for _records, elapsed in results) else: - pipeline = _build_pipeline(data_file, performance_profile, logger) - ExampleUtils.check_data_file(pipeline, logger) - elapsed = _time_lookups(pipeline, evidence) + with _build_pipeline( + data_file, performance_profile, logger) as pipeline: + ExampleUtils.check_data_file(pipeline, logger) + elapsed = _time_lookups(pipeline, evidence) per_second = count / elapsed if elapsed > 0 else 0 ms_per_lookup = (elapsed / count) * 1000 if count > 0 else 0 diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/tests/conftest.py b/ip-intelligence/fiftyone_ipintelligence_examples/tests/conftest.py new file mode 100644 index 00000000..fc53a6d0 --- /dev/null +++ b/ip-intelligence/fiftyone_ipintelligence_examples/tests/conftest.py @@ -0,0 +1,119 @@ +# ********************************************************************* +# This Original Work is copyright of 51 Degrees Mobile Experts Limited. +# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, +# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. +# +# This Original Work is licensed under the European Union Public Licence +# (EUPL) v.1.2 and is subject to its terms as set out below. +# +# If a copy of the EUPL was not distributed with this file, You can obtain +# one at https://opensource.org/licenses/EUPL-1.2. +# +# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be +# amended by the European Commission) shall be deemed incompatible for +# the purposes of the Work and the provisions of the compatibility +# clause in Article 5 of the EUPL shall not apply. +# +# If using the Work as, or as part of, a network application, by +# including the attribution notice(s) required under Article 5 of the EUPL +# in the end user terms of the application under an appropriate heading, +# such notice(s) shall fulfill the requirements of that article. +# ********************************************************************* + +# Fails a test that leaves the native results or engine of an on-premise +# engine alive, because they hold the engine's data in memory. The same +# file is in each test folder that runs on-premise engines. +# +# Reference counting frees a native object as soon as nothing holds it. A +# FlowData refers to itself through its evidence, and a pipeline through +# its elements, so when they are not closed only the cycle collector frees +# what they hold. The collector is off while each test runs, so whatever +# the test leaves open is still alive when it ends. Objects that existed +# before the test, such as a pipeline built in setUpClass, are checked once +# every test has run. + +import gc +import importlib + +import pytest + +# The SWIG modules of the on-premise engines, whichever are installed. +NATIVE_MODULES = [ + "fiftyone_devicedetection_onpremise.DeviceDetectionHashEngineModule", + "fiftyone_ipintelligence_onpremise.IpIntelligenceEngineModule", +] + + +def _native_types(): + types = [] + for name in NATIVE_MODULES: + try: + module = importlib.import_module(name) + except ImportError: + continue + types.append(module.ResultsBaseSwig) + types.append(module.EngineBaseSwig) + return tuple(types) + + +NATIVE_TYPES = _native_types() + +# A failed test is not reported again, because its traceback can hold the +# objects it would have closed. +_failed_tests = set() + +# Objects created when the tests were collected are checked at the end only +# when every test ran, because the teardown that closes them runs with the +# tests. +_some_tests_deselected = False + + +def _alive(): + # The real type is used, because isinstance also asks the object for + # its __class__, which some proxies cannot answer. + return [o for o in gc.get_objects() + if issubclass(type(o), NATIVE_TYPES)] + + +def _describe(names): + return (f"{len(names)} native object(s) still alive " + f"({', '.join(sorted(names))}). Close every FlowData and " + "pipeline the test creates, for example with a with " + "statement, so their native data is released straight away.") + + +def pytest_runtest_logreport(report): + if report.failed: + _failed_tests.add(report.nodeid) + + +def pytest_deselected(items): + global _some_tests_deselected + _some_tests_deselected = True + + +@pytest.fixture(autouse=True) +def native_objects_released(request): + gc.collect() + before = {id(o) for o in _alive()} + gc.disable() + try: + yield + left = [type(o).__name__ for o in _alive() if id(o) not in before] + finally: + gc.enable() + gc.collect() + if left and request.node.nodeid not in _failed_tests: + pytest.fail(_describe(left), pytrace=False) + + +def pytest_sessionfinish(session, exitstatus): + gc.collect() + left = [type(o).__name__ for o in _alive()] + if (left and exitstatus == pytest.ExitCode.OK and + not _some_tests_deselected): + reporter = session.config.pluginmanager.get_plugin("terminalreporter") + if reporter is not None: + reporter.write_line( + "After the last test, " + _describe(left), red=True) + session.exitstatus = pytest.ExitCode.TESTS_FAILED diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/tests/test_examples.py b/ip-intelligence/fiftyone_ipintelligence_examples/tests/test_examples.py index a2d17c30..0ada1f94 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/tests/test_examples.py +++ b/ip-intelligence/fiftyone_ipintelligence_examples/tests/test_examples.py @@ -70,7 +70,9 @@ def test_onpremise_datafile_loads(self): usage_sharing=False, auto_update=False, licence_keys="").build() + self.addCleanup(pipeline.close) data = pipeline.create_flowdata() + self.addCleanup(data.close) data.evidence.add_from_dict({"query.client-ip": "8.8.8.8"}) data.process() # The supplied IP address is echoed back in the results. diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/tox.ini b/ip-intelligence/fiftyone_ipintelligence_examples/tox.ini index 0008718b..3bc4fbd3 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/tox.ini +++ b/ip-intelligence/fiftyone_ipintelligence_examples/tox.ini @@ -24,6 +24,13 @@ deps = # these packages is on a package index yet, so it comes from the # working copy as the others do. -e../fiftyone_ipintelligence_translation/ + # The pipeline packages come from this checkout rather than a package + # index, so the tests run with the pipeline code they are released with. + -e../../pipeline/fiftyone_pipeline_core/ + -e../../pipeline/fiftyone_pipeline_engines/ + -e../../pipeline/fiftyone_pipeline_engines_fiftyone/ + -e../../pipeline/fiftyone_pipeline_cloudrequestengine/ + -e../../pipeline/fiftyone_pipeline_translation/ # --cov is needed even though coverage of the examples is not the point: # common-ci's run-integration-tests.ps1 moves the .coverage file every # package writes, and throws when there is none, after the tests pass. diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/setup.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/setup.py index b15a227a..2d850cf7 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/setup.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/setup.py @@ -277,5 +277,12 @@ def stage_native_source(): py_modules=["fiftyone_ipintelligence_onpremise"], packages=["fiftyone_ipintelligence_onpremise"], package_dir={"": "src"}, - install_requires=["fiftyone_ipintelligence_shared", "fiftyone_pipeline_engines_fiftyone"], + install_requires=[ + "fiftyone_ipintelligence_shared", + "fiftyone_pipeline_engines_fiftyone", + # The first versions with close(), which releases the native memory + # this package holds. + "fiftyone_pipeline_core>=4.5.42", + "fiftyone_pipeline_engines>=4.5.42", + ], ) diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/constants.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/constants.py index 3e129185..deea6331 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/constants.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/constants.py @@ -29,3 +29,10 @@ 'on-premise IP intelligence engine. Its results ' 'read from native memory, which is not safe to ' 'share between requests.') + +ENGINE_CLOSED = ('The on-premise IP intelligence engine has been closed, ' + 'so it cannot process evidence. Build a new pipeline.') + +RESULTS_CLOSED = ('These IP intelligence results have been closed, so ' + 'their properties cannot be read. Read the values needed ' + 'before closing the FlowData.') diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/ipintelligence_onpremise.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/ipintelligence_onpremise.py index c724d64d..37b45904 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/ipintelligence_onpremise.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/ipintelligence_onpremise.py @@ -328,6 +328,20 @@ def on_registration(self, pipeline): super(IpIntelligenceOnPremise, self).on_registration(pipeline) self.init_engine() + def close(self): + + """! + + Stop the data file updates and release the native engine. Its data + is freed once no results read from it are still held, so close + every FlowData the engine processed as well. Closing the pipeline + closes the engine. + + """ + + super(IpIntelligenceOnPremise, self).close() + self.engine = None + def get_client_ips(self, flow_data): """! @@ -379,6 +393,9 @@ def process_internal(self, flow_data): """ + if getattr(self, "engine", None) is None: + raise Exception(ENGINE_CLOSED) + evidence = EvidenceIpiSwig() for key, value in flow_data.evidence.get_all().items(): diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py index ea285c84..40413cf3 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py @@ -25,6 +25,8 @@ from fiftyone_pipeline_engines.missingproperty_service import MissingPropertyService +from .constants import RESULTS_CLOSED + # Number of decimal places used when converting WKB geometry # properties (e.g. Areas) to their WKT string form. WKT_DECIMAL_PLACES = 16 @@ -71,6 +73,33 @@ def __init__(self, flow_element, swig_results, echo_ip=None, echo_ipv6=None): self.missing_property_service = OnPremiseMissingPropertyService() + def close(self): + + """! + Release the native results, which keep the engine's data in memory + while they are held, even after the engine is closed. Values already + read are unaffected, and reading a property afterwards raises an + exception. Closing the FlowData closes its results. + """ + + self.swig_results = None + + def get(self, key): + + """! + Get a property value, once the results are known to be open + @type key: string + @param key: The key of the property to retrieve + @returns mixed + """ + + # Checked here, because AspectData.get turns an exception from + # get_internal into a missing property message. + if self.swig_results is None: + raise Exception(RESULTS_CLOSED) + + return super(SwigData, self).get(key) + @staticmethod def utf8_string_to_str(utf8_string): diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/conftest.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/conftest.py new file mode 100644 index 00000000..fc53a6d0 --- /dev/null +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/conftest.py @@ -0,0 +1,119 @@ +# ********************************************************************* +# This Original Work is copyright of 51 Degrees Mobile Experts Limited. +# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, +# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. +# +# This Original Work is licensed under the European Union Public Licence +# (EUPL) v.1.2 and is subject to its terms as set out below. +# +# If a copy of the EUPL was not distributed with this file, You can obtain +# one at https://opensource.org/licenses/EUPL-1.2. +# +# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be +# amended by the European Commission) shall be deemed incompatible for +# the purposes of the Work and the provisions of the compatibility +# clause in Article 5 of the EUPL shall not apply. +# +# If using the Work as, or as part of, a network application, by +# including the attribution notice(s) required under Article 5 of the EUPL +# in the end user terms of the application under an appropriate heading, +# such notice(s) shall fulfill the requirements of that article. +# ********************************************************************* + +# Fails a test that leaves the native results or engine of an on-premise +# engine alive, because they hold the engine's data in memory. The same +# file is in each test folder that runs on-premise engines. +# +# Reference counting frees a native object as soon as nothing holds it. A +# FlowData refers to itself through its evidence, and a pipeline through +# its elements, so when they are not closed only the cycle collector frees +# what they hold. The collector is off while each test runs, so whatever +# the test leaves open is still alive when it ends. Objects that existed +# before the test, such as a pipeline built in setUpClass, are checked once +# every test has run. + +import gc +import importlib + +import pytest + +# The SWIG modules of the on-premise engines, whichever are installed. +NATIVE_MODULES = [ + "fiftyone_devicedetection_onpremise.DeviceDetectionHashEngineModule", + "fiftyone_ipintelligence_onpremise.IpIntelligenceEngineModule", +] + + +def _native_types(): + types = [] + for name in NATIVE_MODULES: + try: + module = importlib.import_module(name) + except ImportError: + continue + types.append(module.ResultsBaseSwig) + types.append(module.EngineBaseSwig) + return tuple(types) + + +NATIVE_TYPES = _native_types() + +# A failed test is not reported again, because its traceback can hold the +# objects it would have closed. +_failed_tests = set() + +# Objects created when the tests were collected are checked at the end only +# when every test ran, because the teardown that closes them runs with the +# tests. +_some_tests_deselected = False + + +def _alive(): + # The real type is used, because isinstance also asks the object for + # its __class__, which some proxies cannot answer. + return [o for o in gc.get_objects() + if issubclass(type(o), NATIVE_TYPES)] + + +def _describe(names): + return (f"{len(names)} native object(s) still alive " + f"({', '.join(sorted(names))}). Close every FlowData and " + "pipeline the test creates, for example with a with " + "statement, so their native data is released straight away.") + + +def pytest_runtest_logreport(report): + if report.failed: + _failed_tests.add(report.nodeid) + + +def pytest_deselected(items): + global _some_tests_deselected + _some_tests_deselected = True + + +@pytest.fixture(autouse=True) +def native_objects_released(request): + gc.collect() + before = {id(o) for o in _alive()} + gc.disable() + try: + yield + left = [type(o).__name__ for o in _alive() if id(o) not in before] + finally: + gc.enable() + gc.collect() + if left and request.node.nodeid not in _failed_tests: + pytest.fail(_describe(left), pytrace=False) + + +def pytest_sessionfinish(session, exitstatus): + gc.collect() + left = [type(o).__name__ for o in _alive()] + if (left and exitstatus == pytest.ExitCode.OK and + not _some_tests_deselected): + reporter = session.config.pluginmanager.get_plugin("terminalreporter") + if reporter is not None: + reporter.write_line( + "After the last test, " + _describe(left), red=True) + session.exitstatus = pytest.ExitCode.TESTS_FAILED diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_ipintelligence.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_ipintelligence.py index df9f835e..7b14231c 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_ipintelligence.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_ipintelligence.py @@ -49,8 +49,13 @@ def setUpClass(cls): .add(cls.engine) \ .build() + @classmethod + def tearDownClass(cls): + cls.pipeline.close() + def process(self, key, value): flowdata = self.pipeline.create_flowdata() + self.addCleanup(flowdata.close) flowdata.evidence.add(key, value) flowdata.process() return flowdata @@ -133,6 +138,24 @@ def test_process_ipv6(self): ipv6 = flowdata.ip.get("ipv6") self.assertTrue(ipv6.has_value()) + def test_flowdata_close_releases_results(self): + """! + Closing the FlowData releases its native results, and reading a + property afterwards raises an exception. + """ + + with self.pipeline.create_flowdata() as flowdata: + flowdata.evidence.add( + "query.client-ip", TestConstants.IPV4_ADDRESS) + flowdata.process() + registered_name = flowdata.ip.get("registeredname") + + self.assertTrue(registered_name.has_value()) + self.assertIsNone(flowdata.ip.swig_results) + with self.assertRaises(Exception) as context: + flowdata.ip.get("registeredname") + self.assertIn("closed", str(context.exception)) + def test_missing_property(self): """! Accessing a property that does not exist raises an exception @@ -208,10 +231,12 @@ def test_asn_data_file_opens(self): data_file_path=TestConstants.get_asn_data_file_path(), licence_keys="") pipeline = PipelineBuilder().add(engine).build() + self.addCleanup(pipeline.close) self.assertIn("asnname", engine.properties) flowdata = pipeline.create_flowdata() + self.addCleanup(flowdata.close) flowdata.evidence.add("query.client-ip", TestConstants.IPV4_ADDRESS) flowdata.process() @@ -262,3 +287,31 @@ def test_pipeline_builder_refuses_cache(self): cache=LRUEngineCache()).build() self.assertIn("results cache", str(context.exception)) + + +@unittest.skipUnless( + os.path.isfile(data_file), + "No IP intelligence data file found at '%s'. Run ci/fetch-assets.ps1 " + "or set the IPINTELLIGENCEDATAFILE environment variable." % data_file) +class IpIntelligenceCloseTests(unittest.TestCase): + + def test_pipeline_close_releases_engine(self): + """! + Closing the pipeline releases the native engine, and processing + afterwards raises an exception. + """ + + engine = IpIntelligenceOnPremise( + data_file_path=data_file, + licence_keys="") + pipeline = PipelineBuilder().add(engine).build() + + pipeline.close() + + self.assertIsNone(engine.engine) + flowdata = pipeline.create_flowdata() + self.addCleanup(flowdata.close) + flowdata.evidence.add("query.client-ip", TestConstants.IPV4_ADDRESS) + with self.assertRaises(Exception) as context: + flowdata.process() + self.assertIn("closed", str(context.exception)) diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/tox.ini b/ip-intelligence/fiftyone_ipintelligence_onpremise/tox.ini index 3dc3b4c9..211e4571 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/tox.ini +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/tox.ini @@ -24,6 +24,12 @@ deps = # of it in beside the wheel tox builds and made examples and on-premise # depend on each other. -e../fiftyone_ipintelligence_shared/ + # The pipeline packages come from this checkout rather than a package + # index, so the tests run with the pipeline code they are released with. + -e../../pipeline/fiftyone_pipeline_core/ + -e../../pipeline/fiftyone_pipeline_engines/ + -e../../pipeline/fiftyone_pipeline_engines_fiftyone/ + -e../../pipeline/fiftyone_pipeline_cloudrequestengine/ commands = python setup.py build_clib build_ext {[common]commands} diff --git a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowdata.py b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowdata.py index daa6b387..c6f83c7c 100644 --- a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowdata.py +++ b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowdata.py @@ -94,6 +94,34 @@ def process(self): return self + def close(self): + + """! + Release the native resources held by the element data in this + FlowData, such as the results of an on-premise engine, which keep + the engine's data in memory until they are released. Use a with + statement so that this happens as soon as the FlowData is finished + with, rather than when the garbage collector next runs. + + Only element data with a close method of its own is closed, so + element data that needs no cleanup, such as an entry in a results + cache, is left alone. Values already read are unaffected. + + """ + + for element_data in list(self.data.values()): + # Looked up on the class, because ElementData treats an + # attribute it does not have as the name of a property. + close = getattr(type(element_data), "close", None) + if callable(close): + close(element_data) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, exc_traceback): + self.close() + def get_from_element(self, flow_element): """! diff --git a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowelement.py b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowelement.py index 87ee2800..73e6922d 100644 --- a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowelement.py +++ b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/flowelement.py @@ -77,6 +77,17 @@ def on_registration(self, pipeline): pass + def close(self): + + """! + Release the resources the element holds. The pipeline calls this + when it is closed. This element holds none, and elements that hold + native data or threads override it. + + """ + + pass + def get_evidence_key_filter(self): """! diff --git a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipeline.py b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipeline.py index cac6aea1..4ee68108 100644 --- a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipeline.py +++ b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipeline.py @@ -33,7 +33,7 @@ class Pipeline: """ - def __init__(self, flow_elements, logger=Logger(), suppress_process_exceptions = False, data_file_update_service = None): + def __init__(self, flow_elements, logger=Logger(), suppress_process_exceptions = False, data_file_update_service = None, auto_close_elements = True): """! Pipeline constructor. @@ -49,6 +49,12 @@ def __init__(self, flow_elements, logger=Logger(), suppress_process_exceptions = exceptions added to FlowData.errors otherwise will throw the exception occurred during the processing of first element. + @type auto_close_elements: boolean + @param auto_close_elements: If true, the default, closing the + pipeline closes its flow elements. Set it to false when the elements + are shared with another pipeline, and close each element once every + pipeline using it is finished with. + @rtype: Pipeline @return: Returns a Pipeline @@ -63,6 +69,8 @@ def __init__(self, flow_elements, logger=Logger(), suppress_process_exceptions = self.flow_elements_display_list = [] self.suppress_process_exceptions = suppress_process_exceptions + + self.auto_close_elements = auto_close_elements if data_file_update_service != None: self.data_file_update_service = data_file_update_service @@ -92,6 +100,44 @@ def create_flowdata(self): return FlowData(self) + def close(self): + + """! + Close the flow elements, which releases the native data they hold + and stops the updates of their data files, unless the pipeline was + built with auto_close_elements set to false. Use a with statement + so that this happens when the pipeline is finished with. + + FlowData created by the pipeline are not closed by it. Close each + one as well, because the results it holds keep the engine's data in + memory until they are released. + + """ + + if not self.auto_close_elements: + return + + # Every element is closed even if one fails, and the first failure + # is raised once they all have been. + failure = None + for flow_element in self.flow_elements: + close = getattr(flow_element, "close", None) + if close is None: + continue + try: + close() + except Exception as e: + if failure is None: + failure = e + if failure is not None: + raise failure + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, exc_traceback): + self.close() + def log(self, level, message): """! diff --git a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipelinebuilder.py b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipelinebuilder.py index 72f89752..c3fa52e0 100644 --- a/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipelinebuilder.py +++ b/pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/pipelinebuilder.py @@ -52,6 +52,9 @@ def __init__(self, settings= {}): while processing is re-raised from flow_data.process(); if True, errors are stored on flow_data.errors and process() returns normally. Recommended True for web apps so a processing failure degrades gracefully instead of failing the request. + `auto_close_elements (Bool)` - Whether closing the pipeline closes its flow elements + (default true). Set to False when the elements are shared with another pipeline, + and close each element once every pipeline using it is finished with. @return: Returns a Pipeline Builder """ @@ -82,6 +85,11 @@ def __init__(self, settings= {}): else: self.suppress_process_exceptions = False + if "auto_close_elements" in settings: + self.auto_close_elements = settings["auto_close_elements"] + else: + self.auto_close_elements = True + def get_javascript_elements(self): """ Adds the JavaScriptBuilder, JSONBundler and SequenceElement to the pipeline if @@ -148,7 +156,7 @@ def build(self): self.flow_elements.extend(self.get_javascript_elements()) self.flow_elements.extend(self.get_setheader_elements()) - return Pipeline(self.flow_elements, logger=self.logger, suppress_process_exceptions=self.suppress_process_exceptions, data_file_update_service=self.data_file_update_service) + return Pipeline(self.flow_elements, logger=self.logger, suppress_process_exceptions=self.suppress_process_exceptions, data_file_update_service=self.data_file_update_service, auto_close_elements=self.auto_close_elements) def add_logger(self, logger): """ @@ -176,27 +184,46 @@ def build_from_configuration(self, config): @rtype: Pipeline @return: Returns built pipeline + If an element fails to build, the elements already built are closed + and the error is raised. + """ flow_elements = [] - for element in config["PipelineOptions"]["Elements"]: - flow_element = importlib.import_module(element["elementPath"]).__getattribute__(element["elementName"]) - if("elementParameters" in element): - arguments = self._map_properties_names( - { - "verify_md5": "data_update_verify_md5", - "data_file_update_base_url": "data_update_url" - }, - element["elementParameters"], - ) - - flow_element = flow_element(**arguments) - else: - flow_element = flow_element() - flow_elements.append(flow_element) - - return Pipeline(flow_elements, suppress_process_exceptions=self._get_suppress_process_exceptions(config)) + try: + for element in config["PipelineOptions"]["Elements"]: + flow_element = importlib.import_module(element["elementPath"]).__getattribute__(element["elementName"]) + if("elementParameters" in element): + arguments = self._map_properties_names( + { + "verify_md5": "data_update_verify_md5", + "data_file_update_base_url": "data_update_url" + }, + element["elementParameters"], + ) + + flow_element = flow_element(**arguments) + else: + flow_element = flow_element() + flow_elements.append(flow_element) + + return Pipeline(flow_elements, suppress_process_exceptions=self._get_suppress_process_exceptions(config), auto_close_elements=self.auto_close_elements) + except BaseException: + # A failure to close is logged rather than raised, so the caller + # sees the error that stopped the build. + for flow_element in flow_elements: + close = getattr(flow_element, "close", None) + if close is None: + continue + try: + close() + except Exception as e: + self.logger.log( + "error", + "Could not close {} after the pipeline failed to " + "build: {}".format(type(flow_element).__name__, e)) + raise def _get_suppress_process_exceptions(self, config): """ diff --git a/pipeline/fiftyone_pipeline_core/tests/test_close.py b/pipeline/fiftyone_pipeline_core/tests/test_close.py new file mode 100644 index 00000000..e99583d5 --- /dev/null +++ b/pipeline/fiftyone_pipeline_core/tests/test_close.py @@ -0,0 +1,182 @@ +# ********************************************************************* +# This Original Work is copyright of 51 Degrees Mobile Experts Limited. +# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, +# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. +# +# This Original Work is licensed under the European Union Public Licence +# (EUPL) v.1.2 and is subject to its terms as set out below. +# +# If a copy of the EUPL was not distributed with this file, You can obtain +# one at https://opensource.org/licenses/EUPL-1.2. +# +# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be +# amended by the European Commission) shall be deemed incompatible for +# the purposes of the Work and the provisions of the compatibility +# clause in Article 5 of the EUPL shall not apply. +# +# If using the Work as, or as part of, a network application, by +# including the attribution notice(s) required under Article 5 of the EUPL +# in the end user terms of the application under an appropriate heading, +# such notice(s) shall fulfill the requirements of that article. +# ********************************************************************* + +import unittest + +from fiftyone_pipeline_core.elementdata import ElementData +from fiftyone_pipeline_core.elementdata_dictionary import ElementDataDictionary +from fiftyone_pipeline_core.flowelement import FlowElement +from fiftyone_pipeline_core.pipelinebuilder import PipelineBuilder + + +class ClosingData(ElementData): + """Element data with a close method of its own, as the results of an + on-premise engine have.""" + + def __init__(self, flow_element): + super(ClosingData, self).__init__(flow_element) + self.closed = False + + def close(self): + self.closed = True + + +class ClosingElement(FlowElement): + """Adds ClosingData to each FlowData, and counts its own closes.""" + + def __init__(self): + super(ClosingElement, self).__init__() + self.datakey = "closing" + self.closes = 0 + + def process_internal(self, flowdata): + flowdata.set_element_data(ClosingData(self)) + + def close(self): + self.closes += 1 + + +class PlainElement(FlowElement): + """Adds element data that needs no cleanup.""" + + def __init__(self): + super(PlainElement, self).__init__() + self.datakey = "plain" + + def process_internal(self, flowdata): + flowdata.set_element_data(ElementDataDictionary(self, {"close": 1})) + + +class FailingElement(FlowElement): + """Fails to close.""" + + def __init__(self): + super(FailingElement, self).__init__() + self.datakey = "failing" + + def close(self): + raise Exception("close failed") + + +class RecordedElement(FlowElement): + """Records each instance that is closed, for a test whose pipeline + failed to build and so cannot be reached.""" + + closed = [] + + def __init__(self): + super(RecordedElement, self).__init__() + self.datakey = "recorded" + + def close(self): + RecordedElement.closed.append(self) + + +class BrokenElement(FlowElement): + """Fails to build.""" + + def __init__(self): + raise ValueError("element failed to build") + + +def build(*elements, **settings): + settings["add_javascript_builder"] = False + settings["use_setheader_properties"] = False + builder = PipelineBuilder(settings) + for element in elements: + builder.add(element) + return builder.build() + + +class CloseTests(unittest.TestCase): + + # Element data with a close method is closed with the FlowData, and + # element data without one is left alone, even with a property called + # close. + def test_flowdata_closes_element_data(self): + with build(ClosingElement(), PlainElement()) as pipeline: + with pipeline.create_flowdata() as flowdata: + flowdata.process() + closing = flowdata.get("closing") + self.assertFalse(closing.closed) + self.assertTrue(closing.closed) + self.assertEqual(1, flowdata.get("plain").get("close")) + + # The with statement closes the FlowData when an exception is raised. + def test_flowdata_closed_on_exception(self): + with build(ClosingElement()) as pipeline: + with self.assertRaises(ValueError): + with pipeline.create_flowdata() as flowdata: + flowdata.process() + closing = flowdata.get("closing") + raise ValueError() + self.assertTrue(closing.closed) + + # A FlowData that was never processed holds nothing to close. + def test_unprocessed_flowdata_closes(self): + with build(ClosingElement()) as pipeline: + pipeline.create_flowdata().close() + + def test_pipeline_closes_elements(self): + element = ClosingElement() + with build(element): + self.assertEqual(0, element.closes) + self.assertEqual(1, element.closes) + + # Elements shared with another pipeline are left for the caller to + # close. + def test_pipeline_leaves_elements_open(self): + element = ClosingElement() + with build(element, auto_close_elements=False): + pass + self.assertEqual(0, element.closes) + + # Every element is closed even when one fails, and the failure is + # raised afterwards. + def test_pipeline_closes_every_element(self): + element = ClosingElement() + pipeline = build(FailingElement(), element) + with self.assertRaises(Exception) as context: + pipeline.close() + self.assertEqual("close failed", str(context.exception)) + self.assertEqual(1, element.closes) + + # When an element in a configuration fails to build, the elements built + # before it are closed, and the error that stopped the build is raised. + def test_failed_configuration_closes_built_elements(self): + RecordedElement.closed = [] + config = { + "PipelineOptions": { + "Elements": [ + {"elementName": "RecordedElement", "elementPath": __name__}, + {"elementName": "BrokenElement", "elementPath": __name__} + ] + } + } + with self.assertRaises(ValueError) as context: + PipelineBuilder().build_from_configuration(config) + self.assertEqual("element failed to build", str(context.exception)) + self.assertEqual(1, len(RecordedElement.closed)) + + +if __name__ == "__main__": + unittest.main() diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 59168577..c4c6b047 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -90,6 +90,10 @@ def update_data_file(self, data_file): @param data_file: Datafile to update """ + # A data file whose engine has been closed is not updated. + if not self.__is_registered(data_file): + return False + # exit if already updating if data_file.updating: @@ -230,6 +234,9 @@ def check_next_update(self, data_file, run_after_error = False): @param data_file: Datafile to check updates for """ + if not self.__is_registered(data_file): + return + if data_file.update_on_start and not data_file.updated_on_start and not run_after_error: self.update_data_file(data_file) @@ -241,6 +248,7 @@ def check_next_update(self, data_file, run_after_error = False): t = threading.Timer(interval, self.async_update_data_file, args=[data_file]) t.daemon = True t.name = "AutoUpdate - Scheduled" + data_file.update_timer = t t.start() def register_pipeline(self, pipeline): @@ -265,6 +273,26 @@ def register_data_file(self, data_file): data_file.last_changed_time = os.stat(data_file.path).st_mtime self.track_local_data_file_changes(data_file) + def unregister_data_file(self, data_file): + """! + Stop the checks and updates of a datafile, for example when its + engine is closed. Checks already scheduled are cancelled, and a + check already running schedules no other, so nothing is left + holding the data file or its engine. + @type data_file: DataFile + @param data_file: Datafile to stop checking + """ + + data_file.registered = False + for name in ("update_timer", "watcher_timer"): + timer = getattr(data_file, name, None) + if timer is not None: + timer.cancel() + + def __is_registered(self, data_file): + # A data file checked without being registered first is checked as + # before. Only one that has been unregistered is left alone. + return getattr(data_file, "registered", True) def track_local_data_file_changes(self, data_file): """! @@ -272,10 +300,12 @@ def track_local_data_file_changes(self, data_file): the file system watcher. A refresh that fails is logged, and reported with AUTO_UPDATE_REFRESH_FAILED as on the update route. The next check is scheduled whatever happens, so that a later file - is still picked up. + is still picked up, until the data file is unregistered. @type data_file: DataFile @param data_file: Datafile to check updates for """ + if not self.__is_registered(data_file): + return try: changed_time = os.stat(data_file.path).st_mtime except OSError: @@ -296,13 +326,15 @@ def track_local_data_file_changes(self, data_file): "' failed. " + str(e)) self.__emit(UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, data_file) finally: - t = threading.Timer( - self.FILE_SYSTEM_WATCHER_INTERVAL_SECONDS, - self.track_local_data_file_changes, - args=[data_file]) - t.daemon = True - t.name = "AutoUpdate - FileWatcher" - t.start() + if self.__is_registered(data_file): + t = threading.Timer( + self.FILE_SYSTEM_WATCHER_INTERVAL_SECONDS, + self.track_local_data_file_changes, + args=[data_file]) + t.daemon = True + t.name = "AutoUpdate - FileWatcher" + data_file.watcher_timer = t + t.start() def on_complete(self, on_complete_method): self.on_complete_methods.append(on_complete_method) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/engine.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/engine.py index f22f654c..5ef3556a 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/engine.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/engine.py @@ -150,6 +150,22 @@ def on_registration(self, pipeline): pipeline.data_file_update_service = DataFileUpdateService(pipeline) pipeline.data_file_update_service.register_data_file(self.data_file) + def close(self): + + """! + + Stop the checks and updates of the engine's data file, whose timers + would otherwise keep the engine alive. Engines that hold native data + release it as well. + + """ + + if self.data_file: + for pipeline in self.pipelines: + service = getattr(pipeline, "data_file_update_service", None) + if service is not None: + service.unregister_data_file(self.data_file) + def register_data_file(self, data_file): """! diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index af290377..5fd9c872 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -73,10 +73,16 @@ def __init__(self, pending, function, args): self.pending = pending self.function = function self.args = args + self.cancelled = False def start(self): self.pending.append(self) + def cancel(self): + self.cancelled = True + if self in self.pending: + self.pending.remove(self) + class FileSystemWatcherTests(unittest.TestCase): @@ -149,6 +155,39 @@ def testFileSystemWatcherKeepsWatchingWhileFileIsMissing(self): self.assertNotIn( UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, self.statuses) + # Unregistering, as closing the engine does, cancels the next check, and + # a check already running neither refreshes nor schedules another, so + # nothing is left holding the data file. + def testUnregisterStopsFileSystemWatcher(self): + data_file = StandInDataFile(self.path) + self.service.register_data_file(data_file) + check = self.pending[0] + + self.service.unregister_data_file(data_file) + self.assertTrue(check.cancelled) + self.assertEqual(0, len(self.pending)) + + self.write("version 2") + check.function(*check.args) + self.assertEqual("version 1", data_file.data) + self.assertEqual(0, len(self.pending)) + + # Unregistering cancels the next automatic update, and an update that + # starts afterwards does nothing. + def testUnregisterStopsAutomaticUpdates(self): + data_file = StandInDataFile(self.path) + data_file.file_system_watcher = False + data_file.auto_update = True + data_file.polling_interval = 30 + data_file.update_time_maximum_randomisation = 0 + self.service.register_data_file(data_file) + check = self.pending[0] + + self.service.unregister_data_file(data_file) + self.assertTrue(check.cancelled) + self.assertEqual(0, len(self.pending)) + self.assertFalse(self.service.update_data_file(data_file)) + if __name__ == "__main__": unittest.main() diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py index 55ef341f..3173318e 100644 --- a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py +++ b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py @@ -122,6 +122,17 @@ def __init__( def __del__(self): self.thread_pool.shutdown(wait=True) + def close(self): + + """! + Wait for any usage data already being sent, then stop the threads + that send it. The pipeline calls this when it is closed. + + """ + + super(ShareUsage, self).close() + self.thread_pool.shutdown(wait=True) + def get_constant_xml(self): if self.constant_xml is None: coreVersion = version("fiftyone_pipeline_core")