Repository: intake/intake

Issue:
Loading into sql db catalog entry
I read on the SQLAlchemy catalog reader that the entries from the resulting catalog could be loaded by another reader implementing SQLQuery. What’s an example of this & how does that work? I’m asking because I would like to implement a DuckDB catalog reader that makes entries from the information_schema.tables & write to those tables with the results of other readers.

Discussion:
@martindurant: You might find the following test case informative: https://github.com/intake/intake/blob/master/intake/readers/tests/cats/test_sql.py#L85

Here, a SQLite instance is queries by the catalog reader (`reader`). When you `.read()` this, you get a catalog object populated with data entries, the names of the available tables (and views, if you have them). 

I'm not sure how you programmatically write data into duckDB, so if you show how the workflow goes outside of Intake (with a simple way to set up a temporary duck instance), I can probably show you how to encode it in an Intake catalog pipeline.@martindurant: @dprovder , are you still interested in this?@dprovder: Yes absolutely, I will share details very soon@dprovder: ok here's what I've got:

```
import intake
from intake.readers import DuckDB
from intake import Catalog, DataDescription

class DuckDBCatalog(DuckDB):

    imports = {"duckdb"}
    output_instance = "intake.readers.entry:Catalog"
    implements = "intake.readers.datatypes.SQLQuery"
    def _read(self, data, **kwargs):
        """Fetch tables and views from the DuckDB database and return a catalog."""
        import duckdb

        # Use DuckDB's connection logic from parent class
        conn = self._duck(data)
        
        information_schema_tables = intake.readers.DuckSQL(data).read()
        table_names = [name[0] for name in information_schema_tables.table_name.fetchall()]
        
        entries = {
            name: intake.readers.DuckSQL(
                    intake.datatypes.SQLQuery(
                    conn=data.conn,
                    query=f'"{name}"',
                    metadata={"description": f"Entry for {name}", "type": "table"}
                )
            ).to_entry()
            for name in table_names
        }

        return Catalog(entries=entries, metadata=data.metadata)

```

so that for example we can do:

```
catalog = DuckDBCatalog()
data = intake.datatypes.SQLQuery(
    conn={'database':'~/duck_test.db'},
    query="information_schema.tables",
    metadata={"description": "information_schema"}
)
cat = catalog.read(data,metadata={"description": "My DuckDB Catalog"})
entry = cat['mytablename']
entry.read()
```

I guess this works, but the following handles the schema & could easily be modified to accommodate a more detailed query of the information_schema:
```
class DuckTables(SQLQuery):
    """SQLQuery for querying information_schema.tables in a DuckDB database."""

    def __init__(self, conn, schema="main", catalog=None, metadata=None):
        query = f"""
        SELECT * FROM information_schema.tables
        WHERE table_schema = '{schema}'
        """
        if catalog:
            query += f" AND table_catalog = '{catalog}'"
        super().__init__(conn, query, metadata)

catalog = DuckDBCatalog()
data = DuckTables(
    conn={'database':'/Users/danielprovder/pricing/duck_test.db'},
    metadata={"description": "information_schema"}
)
cat = catalog.read(data,metadata={"description": "My DuckDB Catalog"})
cat.entries
```

Also the converter was pretty simple:

```
from intake.readers.convert import BaseConverter
class PandasToDuck(BaseConverter):
    instances = {"pandas:DataFrame": "duckdb:DuckDBPyRelation"}
    func = "duckdb:df"
```

Do you have any suggestions for a converter that would turn a usual catalog into a DuckDBCatalog? The idea would be to write all the entries into duckdb.@martindurant: What you have so far looks very reasonable. Perhaps it can yet be simplified a little - as you say, this feels more like a converter, query->Catalog.

It's funny that "duckdb:df" converts both ways, it is its own inverse.

> The idea would be to write all the entries into duckdb.

This puzzled me: you want to make a table in duckdb of the list of tables in duckdb? That sounds like a pure SQL action where you shouldn't need to round-trip the information via python.@dprovder: Yes, it is funny that .df is its own inverse, I used that to check the implementation 

To clarify on that last bit, I’m looking for a converter that takes a usual catalog, saves all the entries that duckdb can read into a .duck/.db file (or in memory) then returns a DuckCatalog that reads from those saved tables.

For example if I had a catalog with readers/entries from an api, then applying this converter to the catalog would cache the results of the calls in duckdb & the resulting catalog accesses the db instead of executing the api calls.

@martindurant: Oh! So a caching step - I totally hadn't understood that. So a converter catalog->catalog, with side-effects.@dprovder: Yup, you got it. 

I’ve been trying to view the intake abstractions from a functional/category theoretic perspective…it would be nice to have .map/.filter/.reduce on a catalog. I believe V1 had filter if I’m not mistaken.

Of course a monad would be one way to implement side effects. But for clarities sake, it’s probably not the best way to go about things. @martindurant: It had .search, which is similar. You are quite right, .search() on (potentially nested) catalogs makes a lot of sense still, but allowing for itertools-like functional programming makes sense too. We would need to decide just how dict-like a catalog should be.@dprovder: With respect to that last question, I’ll make a new issue a little later just to keep things focused. I have some ideas with respect to that decision which you may find helpful…@martindurant: Sounds good@dprovder: I’m still stuck on how to handle the cacheing though, please don’t forget @dprovder: I took a look at some of the output classes, I realized that the class for cacheing should return a datatype instead of a catalog, which I could then use to pass into a reader for the duckdb catalog as defined above. This is my current progress, I have yet to test it... more updates soon

```
from intake.readers.convert import BaseConverter
from intake.readers.datatypes import SQLQuery
from intake import Catalog
from duckdb import DuckDBPyConnection
from json import dumps
class CatalogToDuckDB(BaseConverter):
    instances = {'intake:Catalog':'intake.readers.datatypes:SQLQuery'}
    imports = {'duckdb','json'}
    
    def run(self,x:Catalog,url:str,storage_options=None, metadata=None,**kwargs):
        conn = DuckDBPyConnection(database=url, read_only=False)
        write = self.cache(conn)
        
        for name, entry in x.items():
            reader = self.get_reader(entry)
            entry_str = dumps(entry._dic_for_comp())
            if reader is None:
                
                continue
            data = reader.read()
            write(name,data,entry_str)
        
        conn.close()
        return SQLQuery(
            conn={'database':url},
            query = 'SELECT * FROM information_schema.tables;',
            storage_options=storage_options,
            metadata=metadata
        )
    
    def cache(conn:DuckDBPyConnection):
        def writer(name,data,comment=None):
            conn.sql(f"CREATE OR REPLACE TABLE '{name}' AS SELECT * FROM '{data}';")
            if comment is not None:
                conn.sql(f"COMMENT ON TABLE '{name}' IS '{comment}';")
        
        return writer

    
    def get_reader(entry):
        entry_data = entry.to_data()
        possible_outputs = entry_data.possible_outputs.values()
        
        if 'duckdb:DuckDBPyRelation' in possible_outputs:
            return entry_data.to_reader(output='duckdb:DuckDBPyRelation')
        if 'pandas:DataFrame' in possible_outputs:
            return entry_data.to_reader(output='pandas:DataFrame')
        else:
            print(f'{name} entry is not Duckable')
            return None
```@martindurant: You are getting ahead of me ....

I think you have all the pieces. What you originally said I think is still true, you need the following:
- a pandas->duckdb converter, which is already in code, above, and should certainly be added to intake
- a catalog->catalog converter, which applies a given thing to each entry of the input; this is the "map" you talked about. If you want to cache immediately and return the cached datasets as entries, then each output entry is a duckdb->pandas reader, and you already know the table name (or construct the SQL to return the table after writing it, maybe). The "map" idea should also be added to intake, and the specific use of it would be an interesting showcase
- if necessary, you can also play with Condition, to decide when the data should be re-cached into duck.

Solution patch:
diff --git a/intake/readers/convert.py b/intake/readers/convert.py
index ddc8644da..ecb9881a3 100644
--- a/intake/readers/convert.py
+++ b/intake/readers/convert.py
@@ -89,8 +89,44 @@ class DuckToPandas(BaseConverter):
     instances = {"duckdb:DuckDBPyRelation": "pandas:DataFrame"}
     func = "duckdb:DuckDBPyConnection.df"
 
-    def run(self, x, *args, **kwargs):
-        return x.df()
+    def run(self, x, *args, with_arrow=True, **kwargs):
+        if with_arrow:
+            import pandas as pd
+
+            table = x.to_arrow_table()
+            data = {
+                col: pd.Series(pd.arrays.ArrowExtensionArray(val))
+                for col, val in zip(table.column_names, table.columns)
+            }
+            return pd.DataFrame(data)
+        else:
+            return x.df()
+
+
+class PandasToDuck(BaseConverter):
+    """Save content of pandas dataframe to Duck internal storage
+
+    Value of ``table`` can be used to point to attached database or
+    output file.
+    """
+
+    instances = {"pandas:DataFrame": "duckdb:DuckDBPyRelation"}
+    func = "duckdb:df"
+
+    def run(self, x, conn: dict | str, table: str, *args, overwrite=True, **kwargs):
+        # TODO: more options like metadata
+        import duckdb
+        from intake.readers.datatypes import SQLQuery
+
+        duck = readers.DuckSQL._duck(None, conn=conn)
+        out = duckdb.df(x, connection=duck)
+        duck.register(view_name="temp_view", python_object=out)
+        duck.sql(
+            f"CREATE {'OR REPLACE' if overwrite else ''} "
+            f"TABLE '{table}' AS SELECT * FROM 'temp_view';"
+        )
+        out = readers.DuckSQL(SQLQuery(conn=conn, query=f"SELECT * FROM '{table}'"))
+        return out
 
 
 class DaskDFToPandas(BaseConverter):
diff --git a/intake/readers/datatypes.py b/intake/readers/datatypes.py
index c5df449a4..ff8bf5653 100644
--- a/intake/readers/datatypes.py
+++ b/intake/readers/datatypes.py
@@ -389,7 +389,7 @@ class OpenDAP(Service):
     structure = {"array"}
 
 
-class SQLQuery(Service):
+class SQLQuery(BaseData):
     """Query on a database-like service"""
 
     structure = {"sequence", "table"}
diff --git a/intake/readers/mixins.py b/intake/readers/mixins.py
index d29d0415d..6db061a51 100644
--- a/intake/readers/mixins.py
+++ b/intake/readers/mixins.py
@@ -29,7 +29,7 @@ def __getattr__(self, item):
         else:
             return out
 
-    def __getitem__(self, item):
+    def __getitem__(self, item: str):
         from intake.readers.convert import Pipeline
         from intake.readers.transform import GetItem
 
@@ -157,7 +157,16 @@ def __getattr__(self, item):
 
         out = [(outtype, func) for outtype, func in self.funcdict.items() if func.__name__ == item]
         if not len(out):
+            # TODO: import class being acted on, to see if attribute requested
+            #  really is available. Perhaps tie to config option.
+            # TODO: exclude certain attributes that might be called during
+            #  a stack trace or other non-normal code, causing accidental
+            #  massive pipelines. E.g., dunders. For those, require `apply()`.
+
             outtype = self.reader.output_instance
+            if item.startswith("_"):
+                raise AttributeError(item)
+
             func = Method
             kw = {"method_name": item}
         else:
diff --git a/intake/readers/readers.py b/intake/readers/readers.py
index 9a46d07e7..38f06c485 100644
--- a/intake/readers/readers.py
+++ b/intake/readers/readers.py
@@ -394,29 +394,34 @@ class DuckDB(BaseReader):
     def discover(self, **kwargs):
         return self.read().limit(10)
 
-    def _duck(self, data):
+    @classmethod
+    def _duck(cls, data, conn=None):
         import duckdb
 
-        conn = getattr(data, "conn", {})  # only SQL type normally has this
-        if isinstance(conn, str):
-            # https://duckdb.org/docs/extensions/
-            if conn.startswith("sqlite:"):
-                duckdb.connect(":default:").execute("INSTALL sqlite;LOAD sqlite;")
-                conn = re.sub("^sqlite3?:/{0,3}", "", conn)
-                conn = {"database": conn}
-            elif conn.startswith("postgres") and str(conn) not in self._dd:
-                d = duckdb.connect()
-                d.execute("INSTALL postgres;LOAD postgres;")
-                # extra params possible here https://duckdb.org/docs/extensions/postgres_scanner#usage
-                d.execute(f"CALL postgres_attach('{conn}');")
-                self._dd[str(conn)] = d
-        if str(conn) not in self._dd:
-            self._dd[str(conn)] = duckdb.connect(
-                **conn
-            )  # connection must be cached for results to be usable
+        conn = getattr(data, "conn", conn or {})  # only SQL type normally has this
+        if str(conn) not in cls._dd:
+            # TODO:  separate engine creation and caching?
+            if isinstance(conn, str):
+                # https://duckdb.org/docs/extensions/
+                if conn.startswith("sqlite:"):
+                    duckdb.connect(":default:").execute("INSTALL sqlite;LOAD sqlite;")
+                    conn1 = re.sub("^sqlite3?:/{0,3}", "", conn)
+                    conn1 = {"database": conn1}
+                    d = duckdb.connect(**conn1)
+                elif conn.startswith("postgres"):
+                    d = duckdb.connect()
+                    d.execute("INSTALL postgres;LOAD postgres;")
+                    # extra params possible here https://duckdb.org/docs/extensions/postgres_scanner#usage
+                    d.execute(f"CALL postgres_attach('{conn}');")
+                else:
+                    d = duckdb.connect(conn)
+            else:
+                d = duckdb.connect(**conn)
+            cls._dd[str(conn)] = d  # connection must be cached for results to be usable
+        d = cls._dd[str(conn)]
         if isinstance(data, datatypes.FileData) and "://" in data.url:
-            self._dd[str(conn)].execute("INSTALL httpfs;LOAD httpfs;")
-        return self._dd[str(conn)]
+            d.execute("INSTALL httpfs;LOAD httpfs;")
+        return d
 
 
 class DuckParquet(DuckDB, FileReader):
diff --git a/intake/readers/transform.py b/intake/readers/transform.py
index 273acd4b6..5337d3ca7 100644
--- a/intake/readers/transform.py
+++ b/intake/readers/transform.py
@@ -2,6 +2,7 @@
 """
 from __future__ import annotations
 
+import intake
 from intake.readers.convert import BaseConverter, SameType
 from intake.readers.utils import one_to_one
 
@@ -108,3 +109,41 @@ class GetItem(BaseConverter):
 
     def _read(self, item, data=None):
         return data[item]
+
+
+class CatalogMapper(BaseConverter):
+    instances = {"intake:Catalog": "intake:Catalog"}
+
+    def __init__(self, func, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.func = func
+        self.args = args
+        self.kwargs = kwargs
+
+    def run(
+        self, in_cat: intake.Catalog, *args, transform=True, name_arg=None, read=False, **kwargs
+    ):
+        """
+        Parameters
+        ----------
+        transform: do we expect this to be a named transform that intake
+            already knows about?
+        name_arg: if give, pass the entry name to the action to be
+            performed using this ad the kwarg name
+        read: if True, execute the pipeline produced. This might be used
+            where the pipeline output is itself another reader.
+        """
+        out = intake.Catalog()
+        for name in in_cat.entries:
+            if name_arg:
+                kwargs[name_arg] = name
+            if transform:
+                pipe = in_cat[name].__getattr__(self.func)(*args, **kwargs)
+            else:
+                func = intake.import_name(self.func)
+                pipe = in_cat[name].apply(func, *args, **kwargs)
+            if read:
+                out[name] = pipe.read()
+            else:
+                out[name] = pipe
+        return out
diff --git a/intake/readers/user_parameters.py b/intake/readers/user_parameters.py
index 31a2df175..9762892b9 100644
--- a/intake/readers/user_parameters.py
+++ b/intake/readers/user_parameters.py
@@ -108,8 +108,32 @@ def _validate(self, value):
         return self.coerce(value) in self.options
 
 
+class NamedOptionsUserParameter(SimpleUserParameter):
+    """One choice out of a given allow dictionary accessed by its keys
+
+    In this case, `dtype` is the types of the dictionary values, and
+    we have a separate "keytype" for the keys, str by default
+    """
+
+    def __init__(self, options, default, dtype=object, keytype=str, **kw):
+        super().__init__(default=options[default], dtype=dtype, **kw)
+        self.keytype = keytype
+        self.options = options
+
+    def coerce(self, value):
+        return self.options.get(value, None)
+
+    def _validate(self, value):
+        if not isinstance(value, self.keytype):
+            value = self.keytype(value)
+        return value in self.options.values()
+
+
 class MultiOptionUserParameter(OptionsUserParameter):
-    """Multiple choices out of a given allow list/tuple"""
+    """Multiple choices out of a given allow list/tuple
+
+    In this case, dtype is the type of each member of the list
+    """
 
     def __init__(self, options: list | tuple, dtype=object, **kw):
         super().__init__(options=options, dtype=dtype, **kw)
diff --git a/scripts/ci/environment-py313.yml b/scripts/ci/environment-py313.yml
index 894b24be1..8eac86a52 100644
--- a/scripts/ci/environment-py313.yml
+++ b/scripts/ci/environment-py313.yml
@@ -39,7 +39,7 @@ dependencies:
   - pre-commit
   - sqlalchemy
   - python-duckdb
-  - pytest-postgresql
+  - pytest-postgresql <7
   - psycopg
   - siphon
   - postgresql


Test patch:
diff --git a/intake/readers/tests/test_up.py b/intake/readers/tests/test_up.py
index 97a0df552..a566fa7ce 100644
--- a/intake/readers/tests/test_up.py
+++ b/intake/readers/tests/test_up.py
@@ -22,3 +22,16 @@ def test_basic():
         # supplied None as a value to int parameter
         pars = {"k": ["{p}", 1], "p": None}
         up.set_values({"p": p}, pars)
+
+
+def test_named_options():
+    from intake.readers import user_parameters as up
+
+    p = up.NamedOptionsUserParameter({"a": "athing", "b": "bthing"}, default="b")
+    pars = {"k": ["{p}", 1]}
+    out = up.set_values({"p": p}, pars)
+    assert out == {"k": ["bthing", 1]}
+
+    pars = {"k": ["{p}", 1], "p": "a"}
+    out = up.set_values({"p": p}, pars)
+    assert out == {"k": ["athing", 1]}
diff --git a/intake/readers/tests/test_workflows.py b/intake/readers/tests/test_workflows.py
index f0821d76a..6459c080d 100644
--- a/intake/readers/tests/test_workflows.py
+++ b/intake/readers/tests/test_workflows.py
@@ -45,6 +45,11 @@ def test_pipelines_in_catalogs(dataframe_file, df):
 def test_pipeline_steps(dataframe_file, df):
     data = intake.readers.datatypes.CSV(url=dataframe_file)
     reader = intake.readers.PandasCSV(data)
+    with pytest.raises(AttributeError):
+        # cannot auto on private methods; could still apply
+        out = reader._nonexistent()
+        breakpoint()
+
     reader2 = reader[["apple", "beet"]].set_index(keys="beet")
     stepper = reader2.read_stepwise()
     assert isinstance(stepper, convert.PipelineExecution)