Adding MLFlow new methods, Julia LTS checks and tests (#28)

* up logbatch function and creating a new type for parameters

* adding julia 1.6 in the CI pipeline

* adding an IOBuffer test

Author: pebeto

.github/workflows/CI.yml

@@ -25,9 +25,8 @@ jobs:
       fail-fast: false
       matrix:
         version:
-#          - '1.0'
-#          - '1.7'
-          - 'nightly'
+          - '1.6'
+          - '1'
         os:
           - ubuntu-latest
         arch:

docs/src/reference.md

@@ -41,6 +41,7 @@ deleterun
 searchruns
 logparam
 logmetric
+logbatch
 logartifact
 listartifacts
 ```

src/MLFlowClient.jl

@@ -30,6 +30,7 @@ export
     MLFlowRunStatus,
     MLFlowRunInfo,
     MLFlowRunDataMetric,
+    MLFlowRunDataParam,
     MLFlowRunData,
     MLFlowRun,
     get_info,
@@ -56,6 +57,7 @@ export
     getexperiment,
     getorcreateexperiment,
     deleteexperiment,
+    restoreexperiment,
     searchexperiments
 
 include("runs.jl")
@@ -68,6 +70,7 @@ export
 
 include("loggers.jl")
 export
+    logbatch,
     logparam,
     logmetric,
     logartifact,

src/experiments.jl

@@ -121,7 +121,7 @@ Deletes an MLFlow experiment.
 function deleteexperiment(mlf::MLFlow, experiment_id::Integer)
     endpoint = "experiments/delete"
     try
-        result = mlfpost(mlf, endpoint; experiment_id=experiment_id)
+        mlfpost(mlf, endpoint; experiment_id=experiment_id)
     catch e
         if isa(e, HTTP.ExceptionRequest.StatusError) && e.status == 404
             # experiment already deleted
@@ -131,6 +131,7 @@ function deleteexperiment(mlf::MLFlow, experiment_id::Integer)
     end
     true
 end
+
 """
     deleteexperiment(mlf::MLFlow, experiment::MLFlowExperiment)
 
@@ -146,6 +147,36 @@ Dispatches to `deleteexperiment(mlf::MLFlow, experiment_id::Integer)`.
 deleteexperiment(mlf::MLFlow, experiment::MLFlowExperiment) =
     deleteexperiment(mlf, experiment.experiment_id)
 
+"""
+    restoreexperiment(mlf::MLFlow, experiment_id::Integer)
+
+Restores a deleted MLFlow experiment.
+
+# Arguments
+- `mlf`: [`MLFlow`](@ref) configuration.
+- `experiment_id`: experiment identifier.
+
+# Returns
+
+`true` if successful. Otherwise, raises exception.
+"""
+function restoreexperiment(mlf::MLFlow, experiment_id::Integer)
+    endpoint = "experiments/restore"
+    try
+        mlfpost(mlf, endpoint; experiment_id=experiment_id)
+    catch e
+        if isa(e, HTTP.ExceptionRequest.StatusError) && e.status == 404
+            # experiment already restored
+            return true
+        end
+        throw(e)
+    end
+    true
+end
+
+restoreexperiment(mlf::MLFlow, experiment::MLFlowExperiment) =
+    restoreexperiment(mlf, experiment.experiment_id)
+
 """
     searchexperiments(mlf::MLFlow)
 

src/loggers.jl

@@ -26,6 +26,7 @@ function logparam(mlf::MLFlow, run::Union{String,MLFlowRun,MLFlowRunInfo}, kv)
     end
 end
 
+
 """
     logmetric(mlf::MLFlow, run, key, value::T; timestamp, step) where T<:Real
     logmetric(mlf::MLFlow, run, key, values::AbstractArray{T}; timestamp, step) where T<:Real
@@ -194,3 +195,37 @@ listartifacts(mlf::MLFlow, run::MLFlowRun; kwargs...) =
     listartifacts(mlf, run.info.run_id; kwargs...)
 listartifacts(mlf::MLFlow, run_info::MLFlowRunInfo; kwargs...) =
     listartifacts(mlf, run_info.run_id; kwargs...)
+
+"""
+    logbatch(mlf::MLFlow, run_id::String, metrics, params, tags)
+
+Logs a batch of metrics, parameters and tags to an experiment run.
+
+# Arguments
+- `mlf::MLFlow`: [`MLFlow`](@ref) onfiguration.
+- `run_id::String`: ID of the run to log to.
+- `metrics`: a vector of [`MLFlowRunDataMetric`](@ref) or a vector of
+NamedTuples of `(name, value, timestamp)`.
+- `params`: a vector of [`MLFlowRunDataParam`](@ref) or a vector of NamedTuples
+of `(name, value)`.
+- `tags`: a vector of strings.
+"""
+logbatch(mlf::MLFlow, run_id::String; tags=String[], metrics=Any[],
+    params=Any[]) = logbatch(mlf, run_id, tags, metrics, params)
+function logbatch(mlf::MLFlow, run_id::String,
+    tags::Union{AbstractVector{<:String}, AbstractVector{Any}},
+    metrics::Union{AbstractVector{<:MLFlowRunDataMetric}, AbstractVector{Any}},
+    params::Union{AbstractVector{<:MLFlowRunDataParam}, AbstractVector{Any}})
+    endpoint = "runs/log-batch"
+    mlfpost(mlf, endpoint;
+        run_id=run_id, metrics=metrics, params=params, tags=tags)
+end
+function logbatch(mlf::MLFlow, run_id::String,
+    tags::Union{AbstractVector{<:String}, AbstractVector{Any}},
+    metrics::Union{AbstractVector{<:AbstractDict}, AbstractVector{Any}},
+    params::Union{AbstractVector{<:AbstractDict}, AbstractVector{Any}})
+    endpoint = "runs/log-batch"
+    mlfpost(mlf, endpoint; run_id=run_id,
+        metrics=MLFlowRunDataMetric.(metrics),
+        params=MLFlowRunDataParam.(params), tags=tags)
+end

src/types/runs.jl

@@ -114,14 +114,39 @@ function MLFlowRunDataMetric(d::Dict{String,Any})
 end
 Base.show(io::IO, t::MLFlowRunDataMetric) = show(io, ShowCase(t, new_lines=true))
 
+"""
+    MLFlowRunDataParam
+
+Represents a parameter.
+
+# Fields
+- `key::String`: parameter identifier.
+- `value::String`: parameter value.
+
+# Constructors
+- `MLFlowRunDataParam(d::Dict{String,String})`
+
+"""
+
+struct MLFlowRunDataParam
+    key::String
+    value::String
+end
+function MLFlowRunDataParam(d::Dict{String,String})
+    key = d["key"]
+    value = d["value"]
+    MLFlowRunDataParam(key, value)
+end
+Base.show(io::IO, t::MLFlowRunDataParam) = show(io, ShowCase(t, new_lines=true))
+
 """
     MLFlowRunData
 
 Represents run data.
 
 # Fields
 - `metrics::Dict{String,MLFlowRunDataMetric}`: run metrics.
-- `params::Dict{String,String}`: run parameters.
+- `params::Dict{String,MLFlowRunDataParam}`: run parameters.
 - `tags`: list of run tags.
 
 # Constructors
@@ -131,24 +156,23 @@ Represents run data.
 """
 struct MLFlowRunData
     metrics::Dict{String,MLFlowRunDataMetric}
-    params::Union{Dict{String,String},Missing}
+    params::Union{Dict{String,MLFlowRunDataParam},Missing}
     tags
 end
 function MLFlowRunData(data::Dict{String,Any})
     metrics = Dict{String,MLFlowRunDataMetric}()
     if haskey(data, "metrics")
         for metric in data["metrics"]
-            v = MLFlowRunDataMetric(metric)
-            metrics[v.key] = v
+            new_metric = MLFlowRunDataMetric(metric)
+            metrics[new_metric.key] = new_metric
         end
     end
+    params = Dict{String,MLFlowRunDataParam}()
     if haskey(data, "params")
-        params = Dict{String,String}()
-        for p in data["params"]
-            params[p["key"]] = p["value"]
+        for param in data["params"]
+            new_param = MLFlowRunDataParam(param["key"], param["value"])
+            params[new_param.key] = new_param
         end
-    else
-        params = Dict{String,String}()
     end
     tags = haskey(data, "tags") ? data["tags"] : missing
     MLFlowRunData(metrics, params, tags)

test/test_experiments.jl

@@ -6,7 +6,7 @@
     deleteexperiment(mlf, exp)
 end
 
-@testset verbose=true "getexperiment" begin
+@testset verbose = true "getexperiment" begin
     @ensuremlf
     exp = createexperiment(mlf)
     experiment = getexperiment(mlf, exp.experiment_id)
@@ -45,13 +45,28 @@ end
 @testset "deleteexperiment" begin
     @ensuremlf
     exp = createexperiment(mlf)
+    deleteexperiment(mlf, exp)
+
+    experiments = searchexperiments(mlf)
+    @test length(experiments) == 1 # 1 for the default experiment
+end
 
-    @test deleteexperiment(mlf, exp)
-    # deleting again to test if the experiment is already deleted
-    @test deleteexperiment(mlf, exp)
+@testset "restoreexperiment" begin
+    @ensuremlf
+    exp = createexperiment(mlf)
+    deleteexperiment(mlf, exp)
+
+    experiments = searchexperiments(mlf)
+    @test length(experiments) == 1 # 1 for the default experiment
+
+    restoreexperiment(mlf, exp)
+    experiments = searchexperiments(mlf)
+    @test length(experiments) == 2 # the restored experiment and the default one
+
+    deleteexperiment(mlf, exp)
 end
 
-@testset verbose=true "searchexperiments" begin
+@testset verbose = true "searchexperiments" begin
     @ensuremlf
     n_experiments = 3
     for i in 2:n_experiments

test/test_loggers.jl

@@ -111,13 +111,55 @@ end
         @test isfile(artifact)
     end
 
+    @testset "logartifact_using_IOBuffer" begin
+        io = IOBuffer()
+        write(io, "testing IOBuffer")
+        seekstart(io)
+        artifact = logartifact(mlf, r, tmpfile, io)
+        @test isfile(artifact)
+    end
+
     @testset "logartifact_error" begin
         @test_broken logartifact(mlf, r, "/etc/shadow")
     end
 
     deleteexperiment(mlf, e)
 end
 
+@testset verbose=true "logbatch" begin
+    @ensuremlf
+    expname = "logbatch-$(UUIDs.uuid4())"
+    e = getorcreateexperiment(mlf, expname)
+    runname = "run-$(UUIDs.uuid4())"
+    r = createrun(mlf, e.experiment_id)
+
+    @testset "logbatch_by_types" begin
+        param_array = [MLFlowRunDataParam("test_param_type", "test")]
+        metric_array = [MLFlowRunDataMetric("test_metric", 5, 3, 1)]
+        logbatch(mlf, r.info.run_id; params=param_array, metrics=metric_array)
+
+        retrieved_run = searchruns(mlf, e;
+            filter_params=Dict("test_param_type" => "test"))
+        @test length(retrieved_run) == 1
+        @test retrieved_run[1].info.run_id == r.info.run_id
+    end
+
+    @testset "logbatch_by_dicts" begin
+        param_dict_array = [Dict("key"=>"test_param_dict", "value"=>"test")]
+        metric_dict_array = [
+            Dict("key"=>"test_metric", "value"=>5, "step"=>3, "timestamp"=>1)]
+        logbatch(mlf, r.info.run_id;
+            params=param_dict_array, metrics=metric_dict_array)
+
+        retrieved_run = searchruns(mlf, e;
+            filter_params=Dict("test_param_dict" => "test"))
+        @test length(retrieved_run) == 1
+        @test retrieved_run[1].info.run_id == r.info.run_id
+    end
+    
+    deleteexperiment(mlf, e)
+end
+
 @testset verbose=true "listartifacts" begin
     @ensuremlf
     expname = "listartifacts-$(UUIDs.uuid4())"