Repository: mlrun/mlrun

Issue:
Add functions to the database
We'd like to store functions in the database. We're going to store YAML/JSON.

@yaronha said:
- YAML, metadata spec, labels
- Key is  project + name + version tag
- put/get/list/delete
- search by project + name + labels


Discussion:

Patch:
diff --git a/examples/mlrun_sparkk8s.ipynb b/examples/mlrun_sparkk8s.ipynb
new file mode 100644
index 000000000..88a0e1aff
--- /dev/null
+++ b/examples/mlrun_sparkk8s.ipynb
@@ -0,0 +1,156 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Execute pyspak script using Spark K8s\n",
+    "This example reads from a MySQL table and writes to an Iguazio KV tables"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!pip install git+https://github.com/mlrun/mlrun\n",
+    "!pip install --upgrade nuclio-jupyter\n",
+    "!pip install kfp"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from mlrun import new_function\n",
+    "import os"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Run script localy (Notebook container)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "user = os.getenv('V3IO_USERNAME','admin')\n",
+    "working_dir  = os.getcwd()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "fn = new_function(command='local://'+ working_dir + '/spark-jdbc.py')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "list_run = fn.run()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Run script on Spark K8S"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Script location.\n",
+    "\n",
+    "Change this location where you cloned mlrun.\n",
+    "\n",
+    "path should be under to /v3io/<container>"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "script_v3io_path = '/v3io/users/admin/sparkk8s'"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "fn = new_function(command='spark://urihoenig/spark-app-25:b2782#local://'+ script_v3io_path +'/spark-jdbc.py')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "fn.with_limits(mem=\"512m\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "fn.spec.wait_for_completion = 1\n",
+    "fn.with_requests(cpu=1)\n",
+    "fn.with_igz_spark('2.5_b2782_20190922162238')\n",
+    "#sj.with_igz_spark()\n",
+    "fn.spec.deps['jars'] += ['/spark/3rd_party/mysql-connector-java-8.0.13.jar']"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "sr = fn.run()"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.6.8"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}
diff --git a/examples/spark-jdbc.py b/examples/spark-jdbc.py
new file mode 100644
index 000000000..e4d996e3c
--- /dev/null
+++ b/examples/spark-jdbc.py
@@ -0,0 +1,22 @@
+# Pyspark example called by mlrun_sparkk8s.ipynb
+import os
+# Iguazio env
+V3IO_USER = os.getenv('V3IO_USERNAME')
+V3IO_HOME = os.getenv('V3IO_HOME')
+V3IO_HOME_URL = os.getenv('V3IO_HOME_URL')
+
+os.environ["PYSPARK_SUBMIT_ARGS"] = "--packages mysql:mysql-connector-java:5.1.39 pyspark-shell"
+
+from pyspark.sql import SparkSession
+
+
+spark = SparkSession.builder.    appName("Spark JDBC to Databases - ipynb").getOrCreate()
+
+
+
+#Loading data from a JDBC source
+dfMySQL = spark.read     .format("jdbc")     .option("url", "jdbc:mysql://mysql-rfam-public.ebi.ac.uk:4497/Rfam")     .option("dbtable", "Rfam.family")     .option("user", "rfamro")     .option("password", "")     .option("driver", "com.mysql.jdbc.Driver")     .load()
+
+
+dfMySQL.write.format("io.iguaz.v3io.spark.sql.kv").mode("overwrite").option("key", "rfam_id").save("v3io://users/admin/frommysql")
+
diff --git a/mlrun/db/base.py b/mlrun/db/base.py
index c1ea83750..be3ac8791 100644
--- a/mlrun/db/base.py
+++ b/mlrun/db/base.py
@@ -89,3 +89,15 @@ class RunDBInterface(ABC):
 
     def read_metric(self, keys, project='', query=''):
         warnings.warn('store_metric not implemented yet')
+
+    @abstractmethod
+    def store_function(self, func, name, project='', tag=''):
+        pass
+
+    @abstractmethod
+    def get_function(self, name, project='', tag=''):
+        pass
+
+    @abstractmethod
+    def list_functions(self, name, project='', tag='', labels=None):
+        pass
diff --git a/mlrun/db/filedb.py b/mlrun/db/filedb.py
index 665c51d22..bdcd9ed58 100644
--- a/mlrun/db/filedb.py
+++ b/mlrun/db/filedb.py
@@ -21,9 +21,14 @@ from datetime import datetime, timedelta
 
 from ..utils import get_in, match_labels, dict_to_yaml, update_in, dict_to_json
 from ..datastore import StoreManager
-from ..render import run_to_html
 from .base import RunDBError, RunDBInterface
-from ..lists import RunList, ArtifactList
+from ..lists import RunList, ArtifactList, FunctionList
+from ..utils import logger
+
+run_logs = 'runs'
+artifacts_dir = 'artifacts'
+functions_dir = 'functions'
+_missing = object()
 
 
 class FileRunDB(RunDBInterface):
@@ -41,11 +46,12 @@ class FileRunDB(RunDBInterface):
         return self
 
     def store_log(self, uid, project='', body=None, append=True):
-        filepath = self._filepath('runs', project, uid, '') + '.log'
-        self._datastore.put(filepath, body, append=append)
+        filepath = self._filepath(run_logs, project, uid, '') + '.log'
+        # TODO: handle append
+        self._datastore.put(filepath, body)
 
     def get_log(self, uid, project='', offset=0):
-        filepath = self._filepath('runs', project, uid, '') + '.log'
+        filepath = self._filepath(run_logs, project, uid, '') + '.log'
         if pathlib.Path(filepath).is_file():
             with open(filepath, 'rb') as fp:
                 if offset:
@@ -54,11 +60,8 @@ class FileRunDB(RunDBInterface):
         return None
 
     def store_run(self, struct, uid, project='', commit=False):
-        if self.format == '.yaml':
-            data = dict_to_yaml(struct)
-        else:
-            data = dict_to_json(struct)
-        filepath = self._filepath('runs', project, uid, '') + self.format
+        data = self._dumps(struct)
+        filepath = self._filepath(run_logs, project, uid, '') + self.format
         self._datastore.put(filepath, data)
 
     def update_run(self, updates: dict, uid, project=''):
@@ -69,21 +72,21 @@ class FileRunDB(RunDBInterface):
         self.store_run(run, uid, project, True)
 
     def read_run(self, uid, project=''):
-        filepath = self._filepath('runs', project, uid, '') + self.format
+        filepath = self._filepath(run_logs, project, uid, '') + self.format
         data = self._datastore.get(filepath)
         return self._loads(data)
 
     def list_runs(self, name='', uid=None, project='', labels=[],
                   state='', sort=True, last=30):
-        filepath = self._filepath('runs', project)
+        filepath = self._filepath(run_logs, project)
         results = RunList()
         if isinstance(labels, str):
             labels = labels.split(',')
         for run, _ in self._load_list(filepath, '*'):
-            if (name == '' or name in get_in(run, 'metadata.name', ''))\
-                    and match_labels(get_in(run, 'metadata.labels', {}), labels)\
-                    and (state == '' or get_in(run, 'status.state', '') == state)\
-                    and (not uid or uid == get_in(run, 'metadata.uid', '')):
+            if match_value(name, run, 'metadata.name') and \
+               match_labels(get_in(run, 'metadata.labels', {}), labels) and \
+               match_value(state, run, 'status.state') and \
+               match_value(uid, run, 'metadata.uid'):
                 results.append(run)
 
         if sort or last:
@@ -94,14 +97,15 @@ class FileRunDB(RunDBInterface):
         return results
 
     def del_run(self, uid, project=''):
-        filepath = self._filepath('runs', project, uid, '') + self.format
+        filepath = self._filepath(run_logs, project, uid, '') + self.format
         self._safe_del(filepath)
 
     def del_runs(self, name='', project='', labels=[], state='', days_ago=0):
         if not name and not state and not days_ago:
-            raise RunDBError('filter is too wide, select name and/or state and/or days_ago')
+            raise RunDBError(
+                'filter is too wide, select name and/or state and/or days_ago')
 
-        filepath = self._filepath('runs', project)
+        filepath = self._filepath(run_logs, project)
         if isinstance(labels, str):
             labels = labels.split(',')
 
@@ -113,33 +117,33 @@ class FileRunDB(RunDBInterface):
                                      '%Y-%m-%d %H:%M:%S.%f') < days_ago
 
         for run, p in self._load_list(filepath, '*'):
-            if (name == '' or name == get_in(run, 'metadata.name', ''))\
-                    and match_labels(get_in(run, 'metadata.labels', {}), labels)\
-                    and (state == '' or get_in(run, 'status.state', '') == state)\
-                    and (not days_ago or date_before(run)):
-
+            if match_value(name, run, 'metadata.name') and \
+               match_labels(get_in(run, 'metadata.labels', {}), labels) and \
+               match_value(state, run, 'status.state') and \
+               (not days_ago or date_before(run)):
                 self._safe_del(p)
 
     def store_artifact(self, key, artifact, uid, tag='', project=''):
         artifact['updated'] = time.time()
-        if self.format == '.yaml':
-            data = dict_to_yaml(artifact)
-        else:
-            data = dict_to_json(artifact)
-        filepath = self._filepath('artifacts', project, key, uid) + self.format
+        data = self._dumps(artifact)
+        filepath = self._filepath(
+            artifacts_dir, project, key, uid) + self.format
         self._datastore.put(filepath, data)
-        filepath = self._filepath('artifacts', project, key, tag or 'latest') + self.format
+        filepath = self._filepath(
+            artifacts_dir, project, key, tag or 'latest') + self.format
         self._datastore.put(filepath, data)
 
     def read_artifact(self, key, tag='', project=''):
-        filepath = self._filepath('artifacts', project, key, tag) + self.format
+        filepath = self._filepath(
+            artifacts_dir, project, key, tag) + self.format
         data = self._datastore.get(filepath)
         return self._loads(data)
 
     def list_artifacts(self, name='', project='', tag='', labels=[]):
         tag = tag or 'latest'
-        print(f'reading artifacts in {project} name/mask: {name} tag: {tag} ...')
-        filepath = self._filepath('artifacts', project, tag=tag)
+        logger.info(
+            f'reading artifacts in {project} name/mask: {name} tag: {tag} ...')
+        filepath = self._filepath(artifacts_dir, project, tag=tag)
         results = ArtifactList()
         results.tag = tag
         if isinstance(labels, str):
@@ -160,12 +164,13 @@ class FileRunDB(RunDBInterface):
         return results
 
     def del_artifact(self, key, tag='', project=''):
-        filepath = self._filepath('artifacts', project, key, tag) + self.format
+        filepath = self._filepath(
+            artifacts_dir, project, key, tag) + self.format
         self._safe_del(filepath)
 
     def del_artifacts(self, name='', project='', tag='', labels=[]):
         tag = tag or 'latest'
-        filepath = self._filepath('artifacts', project, tag=tag)
+        filepath = self._filepath(artifacts_dir, project, tag=tag)
 
         if isinstance(labels, str):
             labels = labels.split(',')
@@ -182,21 +187,70 @@ class FileRunDB(RunDBInterface):
 
                 self._safe_del(p)
 
+    def store_function(self, func, name, project='', tag=''):
+        data = self._dumps(func)
+        filepath = self._filepath(functions_dir, project, name) + self.format
+        self._datastore.put(filepath, data)
+        filepath = self._filepath(
+            functions_dir, project, name, tag or 'latest') + self.format
+        self._datastore.put(filepath, data)
+
+    def get_function(self, name, project='', tag=''):
+        filepath = self._filepath(
+            functions_dir, project, name, tag) + self.format
+        data = self._datastore.get(filepath)
+        return self._loads(data)
+
+    def list_functions(self, name, project='', tag='', labels=None):
+        tag = tag or 'latest'
+        labels = labels or []
+        logger.info(
+            f'reading functions in {project} name/mask: {name} tag: {tag} ...')
+        filepath = self._filepath(functions_dir, project, tag=tag)
+        results = FunctionList(tag)
+        if isinstance(labels, str):
+            labels = labels.split(',')
+        if tag == '*':
+            mask = '**/*' + name
+            if name:
+                mask += '*'
+        else:
+            mask = '**/*'
+        for func, p in self._load_list(filepath, mask):
+            if (name == '' or name in get_in(func, 'name', ''))\
+                    and match_labels(get_in(func, 'labels', {}), labels):
+                if 'artifacts/latest' in p:
+                    func['tree'] = 'latest'
+                results.append(func)
+
+        return results
+
     def _filepath(self, table, project, key='', tag=''):
         if tag == '*':
             tag = ''
         if tag:
             key = '/' + key
         if project:
-            return path.join(self.dirpath, '{}/{}/{}{}'.format(table, project, tag, key))
+            return path.join(self.dirpath, '{}/{}/{}{}'.format(
+                table, project, tag, key))
         else:
             return path.join(self.dirpath, '{}/{}{}'.format(table, tag, key))
 
+    _encodings = {
+        '.yaml': ('to_yaml', dict_to_yaml),
+        '.json': ('to_json', dict_to_json),
+    }
+
     def _dumps(self, obj):
-        if self.format == '.yaml':
-            return obj.to_yaml()
-        else:
-            return obj.to_json()
+        meth_name, enc_fn = self._encodings.get(self.format, (None, None))
+        if meth_name is None:
+            raise ValueError(f'unsupported format - {self.format}')
+
+        meth = getattr(obj, meth_name, None)
+        if meth:
+            return meth()
+
+        return enc_fn(obj)
 
     def _loads(self, data):
         if self.format == '.yaml':
@@ -218,3 +272,9 @@ class FileRunDB(RunDBInterface):
             remove(filepath)
         else:
             raise RunDBError(f'run file is not found or valid ({filepath})')
+
+
+def match_value(value, obj, key):
+    if not value:
+        return True
+    return get_in(obj, key, _missing) == value
diff --git a/mlrun/db/httpd.py b/mlrun/db/httpd.py
index ac6e66976..e7942f9fa 100644
--- a/mlrun/db/httpd.py
+++ b/mlrun/db/httpd.py
@@ -294,6 +294,46 @@ def del_artifacts():
     _file_db.del_artifacts(name, project, tag, labels)
     return jsonify(ok=True)
 
+# curl -d@/path/to/func.json http://localhost:8080/func/prj/7?tag=0.3.2
+@app.route('/func/<project>/<name>', methods=['POST'])
+@catch_err
+def store_function(project, name):
+    try:
+        data = request.get_json(force=True)
+    except ValueError:
+        return json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
+
+    tag = request.args.get('tag', '')
+
+    _file_db.store_function(data, name, project, tag)
+    return jsonify(ok=True)
+
+
+# curl http://localhost:8080/log/prj/7?tag=0.2.3
+@app.route('/func/<project>/<name>', methods=['GET'])
+@catch_err
+def get_function(project, name):
+    tag = request.args.get('tag', '')
+    func = _file_db.get_function(name, project, tag)
+    return jsonify(ok=True, func=func)
+
+
+# curl http://localhost:8080/funcs?project=p1&name=x&label=l1&label=l2
+@app.route('/funcs', methods=['GET'])
+@catch_err
+def list_functions():
+    name = request.args.get('name', '')
+    project = request.args.get('project', 'default')
+    tag = request.args.get('tag', '')
+    labels = request.args.getlist('label')
+
+    out = _file_db.list_functions(name, project, tag, labels)
+    return jsonify(
+        ok=True,
+        funcs=list(out),
+        tag=out.tag,
+    )
+
 
 @app.route('/healthz', methods=['GET'])
 def health():
diff --git a/mlrun/db/httpdb.py b/mlrun/db/httpdb.py
index 75d6210b3..8bc7b349f 100644
--- a/mlrun/db/httpdb.py
+++ b/mlrun/db/httpdb.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
+
 import requests
 
 from ..utils import dict_to_json
@@ -197,6 +199,34 @@ class HTTPRunDB(RunDBInterface):
         error = 'del artifacts'
         self._api_call('DELETE', 'artifacts', error, params=params)
 
+    def store_function(self, func, name, project='', tag=''):
+        params = {'tag': tag}
+        project = project or default_project
+        path = self._path_of('func', project, name)
+
+        error = f'store function {project}/{name}'
+        self._api_call(
+            'POST', path, error, params=params, body=json.dumps(func))
+
+    def get_function(self, name, project='', tag=''):
+        params = {'tag': tag}
+        project = project or default_project
+        path = self._path_of('func', project, name)
+        error = f'get function {project}/{name}'
+        resp = self._api_call('GET', path, error, params=params)
+        return resp.json()['func']
+
+    def list_functions(self, name, project='', tag='', labels=None):
+        params = {
+            'project': project or default_project,
+            'name': name,
+            'tag': tag,
+            'label': labels or [],
+        }
+        error = f'list functions'
+        resp = self._api_call('GET', 'funcs', error, params=params)
+        return resp.json()['funcs']
+
 
 def _as_json(obj):
     fn = getattr(obj, 'to_json', None)
diff --git a/mlrun/lists.py b/mlrun/lists.py
index 138bd8f37..c50904685 100644
--- a/mlrun/lists.py
+++ b/mlrun/lists.py
@@ -80,3 +80,9 @@ class ArtifactList(list):
         html = artifacts_to_html(df, display, classes=classes)
         if not display:
             return html
+
+
+class FunctionList(list):
+    def __init__(self, tag):
+        self.tag = tag
+        # TODO
diff --git a/mlrun/runtimes/sparkjob.py b/mlrun/runtimes/sparkjob.py
index 996625dda..8e8e2e0a0 100644
--- a/mlrun/runtimes/sparkjob.py
+++ b/mlrun/runtimes/sparkjob.py
@@ -11,11 +11,12 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
+import json
 from pprint import pprint
 import yaml
 import time
 from copy import deepcopy
+from .base import RunError
 from .kubejob import KubejobRuntime, KubejobSpec
 from ..model import ModelObj
 from ..utils import logger, get_in
@@ -97,6 +98,9 @@ class SparkJobSpec(KubejobSpec):
         self.spark_version = spark_version
         self.restart_policy = restart_policy
         self.deps = deps
+        self.wait_for_completion = 0   # Seconds to wait for job to complete 0 = no wait
+        self.job_check_interval = 10   # Check job status every N seconds (Only relevant if wait_for completion is set)
+        self.wait_timeout = 0          # Wait N seconds before killing the job
 
 class SparkRuntime(KubejobRuntime):
     group = 'sparkoperator.k8s.io'
@@ -108,7 +112,6 @@ class SparkRuntime(KubejobRuntime):
     def _run(self, runobj: RunObject, execution):
         job = deepcopy(_sparkjob_template)
         meta = self._get_meta(runobj, True)
-
         pod_labels = deepcopy(meta.labels)
         pod_labels['mlrun/job'] = meta.name
         update_in(job, 'metadata', meta.to_dict())
@@ -134,7 +137,35 @@ class SparkRuntime(KubejobRuntime):
         if self.spec.command:
             update_in(job, 'spec.mainApplicationFile', self.spec.command)
         update_in(job, 'spec.args', self.spec.args)
-        self._submit_job(job, meta.namespace)
+        resp = self._submit_job(job, meta.namespace)
+        name = get_in(resp, 'metadata.name', 'unknown')
+        job_response = resp
+        # If wait is set
+        if self.spec.wait_for_completion > 0:
+            import time
+
+            running = "STARTING"
+            job_name = get_in(job_response, 'metadata.name', 'unknown')
+            logger.info('Waiting for application to start')
+            while running not in ["RUNNING", "COMPLETED", "FAILED"]:
+                result = self._get_job_status(namespace=meta.namespace, job_name=job_name)
+                if 'status' in result:
+                    running = result['status']['applicationState']['state']
+                time.sleep(self.spec.job_check_interval)
+
+            logger.info('Application status:' + running)
+            if running == "FAILED":
+                raise RunError('Failed to execute application')
+
+            logger.info('Waiting for application to complete')
+            while running not in ["COMPLETED", "FAILED"]:
+                result = self._get_job_status(namespace=meta.namespace, job_name=job_name)
+                running = result['status']['applicationState']['state']
+                time.sleep(self.spec.job_check_interval)
+
+            if running == "FAILED":
+                raise RunError('Execution failed check the pod logs')
+            logger.info('Application completed:')
 
     def _submit_job(self, job, namespace=None):
         k8s = self._get_k8s()
@@ -149,6 +180,17 @@ class SparkRuntime(KubejobRuntime):
         except client.rest.ApiException as e:
             logger.error("Exception when creating SparkJob: %s" % e)
 
+    def _get_job_status(self, namespace=None, job_name=None):
+        k8s = self._get_k8s()
+        namespace = k8s.ns(namespace)
+        try:
+            resp = k8s.crdapi.get_namespaced_custom_object(
+                SparkRuntime.group, SparkRuntime.version, namespace, SparkRuntime.plural, job_name)
+        except client.rest.ApiException as e:
+            print("Exception when reading SparkJob: %s" % e)
+        return resp
+
+
     def _update_igz_jars(self, igz_version, deps=igz_deps):
         if not self.spec.deps:
             self.spec.deps = {}


Test patch:
diff --git a/tests/test_filedb.py b/tests/test_filedb.py
new file mode 100644
index 000000000..6fc42910b
--- /dev/null
+++ b/tests/test_filedb.py
@@ -0,0 +1,33 @@
+from tempfile import mkdtemp
+
+import pytest
+
+from mlrun.db import FileRunDB
+
+
+@pytest.fixture
+def db():
+    path = mkdtemp(prefix='mlrun-test')
+    db = FileRunDB(dirpath=path)
+    db.connect()
+    return db
+
+
+def test_save_get_function(db: FileRunDB):
+    func, name, proj = {'x': 1, 'y': 2}, 'f1', 'p2'
+    db.store_function(func, name, proj)
+    db_func = db.get_function(name, proj)
+    assert db_func == func, 'wrong func'
+
+
+def test_list_fuctions(db: FileRunDB):
+    proj = 'p4'
+    count = 5
+    for i in range(count):
+        name = f'func{i}'
+        func = {'fid': i}
+        db.store_function(func, name, proj)
+    db.store_function({}, 'f2', 'p7')
+
+    out = db.list_functions('', proj)
+    assert len(out) == count, 'bad list'
diff --git a/tests/test_httpdb.py b/tests/test_httpdb.py
index 88b04ccd7..32daac4a3 100644
--- a/tests/test_httpdb.py
+++ b/tests/test_httpdb.py
@@ -209,3 +209,29 @@ def test_bearer_auth(create_server):
 
     db.token = token
     db.list_runs()
+
+
+def test_set_get_function(create_server):
+    server: Server = create_server()
+    db: HTTPRunDB = server.conn
+
+    func, name, proj = {'x': 1, 'y': 2}, 'f1', 'p2'
+    db.store_function(func, name, proj)
+    db_func = db.get_function(name, proj)
+    assert db_func == func, 'wrong func'
+
+
+def test_list_functions(create_server):
+    server: Server = create_server()
+    db: HTTPRunDB = server.conn
+
+    proj = 'p4'
+    count = 5
+    for i in range(count):
+        name = f'func{i}'
+        func = {'fid': i}
+        db.store_function(func, name, proj)
+    db.store_function({}, 'f2', 'p7')
+
+    out = db.list_functions('', proj)
+    assert len(out) == count, 'bad list'