feat(native): cube.py - support file_repository (#7107)

Author: ovr

packages/cubejs-backend-native/Cargo.lock

@@ -311,6 +311,7 @@ export const shutdownInterface = async (instance: SqlInterfaceInstance): Promise
 };
 
 export interface PyConfiguration {
+    repositoryFactory?: (ctx: unknown) => Promise<unknown>,
     checkAuth?: (req: unknown, authorization: string) => Promise<void>
     queryRewrite?: (query: unknown, ctx: unknown) => Promise<unknown>
     contextToApiScopes?: () => Promise<string[]>
@@ -337,6 +338,15 @@ export const pythonLoadConfig = async (content: string, options: { fileName: str
     );
   }
 
+  if (config.repositoryFactory) {
+    const nativeRepositoryFactory = config.repositoryFactory;
+    config.repositoryFactory = (ctx: any) => ({
+      dataSchemaFiles: async () => nativeRepositoryFactory(
+        ctx
+      )
+    });
+  }
+
   return config;
 };
 

packages/cubejs-backend-native/js/index.ts

@@ -1,6 +1,27 @@
+import os
 from typing import Union, Callable, Dict
 
 
+def file_repository(path):
+    files = []
+
+    for (dirpath, dirnames, filenames) in os.walk(path):
+        for fileName in filenames:
+            if fileName.endswith(".js") or fileName.endswith(".yml") or fileName.endswith(".yaml") or fileName.endswith(".jinja") or fileName.endswith(".py"):
+                path = os.path.join(dirpath, fileName)
+
+                f = open(path, 'r')
+                content = f.read()
+                f.close()
+
+                files.append({
+                    'fileName': fileName,
+                    'content': content
+                })
+
+    return files
+
+
 class RequestContext:
     url: str
     method: str
@@ -29,6 +50,7 @@ class Configuration:
     extend_context: Callable
     scheduled_refresh_contexts: Callable
     context_to_api_scopes: Callable
+    repository_factory: Callable
 
     def __init__(self):
         self.schema_path = None
@@ -53,6 +75,7 @@ def __init__(self):
         self.extend_context = None
         self.scheduled_refresh_contexts = None
         self.context_to_api_scopes = None
+        self.repository_factory = None
 
     def set_schema_path(self, schema_path: str):
         self.schema_path = schema_path
@@ -114,5 +137,8 @@ def set_extend_context(self, extend_context: Callable[[RequestContext], Dict]):
     def set_scheduled_refresh_contexts(self, scheduled_refresh_contexts: Callable):
         self.scheduled_refresh_contexts = scheduled_refresh_contexts
 
+    def set_repository_factory(self, repository_factory: Callable):
+        self.repository_factory = repository_factory
+
 
 settings = Configuration()

packages/cubejs-backend-native/python/cube/src/conf/__init__.py

@@ -4,7 +4,7 @@ use neon::prelude::*;
 use neon::result::Throw;
 use neon::types::JsDate;
 use pyo3::exceptions::{PyNotImplementedError, PyTypeError};
-use pyo3::types::{PyBool, PyDict, PyFloat, PyFunction, PyInt, PyList, PyString};
+use pyo3::types::{PyBool, PyDict, PyFloat, PyFunction, PyInt, PyList, PySet, PyString};
 use pyo3::{Py, PyAny, PyErr, PyObject, Python, ToPyObject};
 use std::cell::RefCell;
 use std::collections::hash_map::{IntoIter, Iter, Keys};
@@ -264,6 +264,15 @@ impl CLRepr {
                     r.push(Self::from_python_ref(v)?);
                 }
 
+                Self::Array(r)
+            } else if v.get_type().is_subclass_of::<PySet>()? {
+                let l = v.downcast::<PySet>()?;
+                let mut r = Vec::with_capacity(l.len());
+
+                for v in l.iter() {
+                    r.push(Self::from_python_ref(v)?);
+                }
+
                 Self::Array(r)
             } else {
                 return Err(PyErr::new::<PyTypeError, _>(format!(

packages/cubejs-backend-native/src/python/cross.rs

@@ -44,6 +44,7 @@ impl CubeConfigPy {
         self.function_attr(config_module, "extend_context")?;
         self.function_attr(config_module, "scheduled_refresh_contexts")?;
         self.function_attr(config_module, "context_to_api_scopes")?;
+        self.function_attr(config_module, "repository_factory")?;
 
         Ok(())
     }

packages/cubejs-backend-native/src/python/cube_config.rs

@@ -1,4 +1,7 @@
-from cube.conf import settings
+from cube.conf import (
+    settings,
+    file_repository
+)
 
 settings.schema_path = "models"
 settings.pg_sql_port = 5555
@@ -16,6 +19,13 @@ async def check_auth(req, authorization):
 
 settings.check_auth = check_auth
 
+async def repository_factory(ctx):
+    print('[python] repository_factory ctx=', ctx)
+
+    return file_repository(ctx['securityContext']['schemaPath'])
+
+settings.repository_factory = repository_factory
+
 async def context_to_api_scopes():
     print('[python] context_to_api_scopes')
     return ['meta', 'data', 'jobs']

packages/cubejs-backend-native/test/config.py

@@ -0,0 +1,3 @@
+cubes:
+  - name: cube_01
+    sql_table: 'kek'

packages/cubejs-backend-native/test/fixtures/schema-tenant-1/test.yml

@@ -0,0 +1,28 @@
+{%- set payment_methods = [
+  "bank_transfer",
+  "credit_card",
+  "gift_card"
+] -%}
+
+cubes:
+  - name: cube_01_1
+    sql: >
+      SELECT
+        order_id,
+        {%- for payment_method in payment_methods %}
+        SUM(CASE WHEN payment_method = '{{payment_method}}' THEN amount END) AS {{payment_method}}_amount,
+        {%- endfor %}
+        SUM(amount) AS total_amount
+      FROM app_data.payments
+      GROUP BY 1
+      
+  - name: cube_01_2
+    sql: >
+      SELECT
+        order_id,
+        {%- for payment_method in payment_methods %}
+        SUM(CASE WHEN payment_method = '{{payment_method}}' THEN amount END) AS {{payment_method}}_amount
+        {%- if not loop.last %},{% endif %}
+        {%- endfor %}
+      FROM app_data.payments
+      GROUP BY 1

packages/cubejs-backend-native/test/fixtures/schema-tenant-1/test.yml.jinja

@@ -42,6 +42,7 @@ suite('Python Config', () => {
       contextToApiScopes: expect.any(Function),
       checkAuth: expect.any(Function),
       queryRewrite: expect.any(Function),
+      repositoryFactory: expect.any(Function),
     });
 
     if (!config.checkAuth) {
@@ -62,6 +63,31 @@ suite('Python Config', () => {
     expect(await config.contextToApiScopes()).toEqual(['meta', 'data', 'jobs']);
   });
 
+  test('repository factory', async () => {
+    if (!config.repositoryFactory) {
+      throw new Error('repositoryFactory was not defined in config.py');
+    }
+
+    const ctx = {
+      securityContext: { schemaPath: path.join(process.cwd(), 'test', 'fixtures', 'schema-tenant-1') }
+    };
+
+    const repository: any = await config.repositoryFactory(ctx);
+    expect(repository).toEqual({
+      dataSchemaFiles: expect.any(Function)
+    });
+
+    const files = await repository.dataSchemaFiles();
+    expect(files).toContainEqual({
+      fileName: 'test.yml',
+      content: expect.any(String),
+    });
+    expect(files).toContainEqual({
+      fileName: 'test.yml.jinja',
+      content: expect.any(String),
+    });
+  });
+
   test('cross language converting (js -> python -> js)', async () => {
     if (!config.queryRewrite) {
       throw new Error('queryRewrite was not defined in config.py');
@@ -81,6 +107,11 @@ suite('Python Config', () => {
       obj: {
         field_str: 'string',
       },
+      obj_with_nested_object: {
+        sub_object: {
+          sub_field_str: 'string'
+        }
+      },
       array_int: [1, 2, 3, 4, 5],
       array_obj: [{
         field_str_first: 'string',