feat: add a duckdb handle with the project filesystem registered unde…

…r cdf protocol

src/cdf/core/filesystem.py

@@ -52,7 +52,8 @@ def wrapped(self) -> fsspec.AbstractFileSystem:
 
         options = self.options.copy()
         options.setdefault("auto_mkdir", True)
-        return DirFileSystem(
+        CdfFs = type("CdfFs", (DirFileSystem,), {"protocol": "cdf"})
+        return CdfFs(
             path=posixpath.join(strip_protocol(self.uri), "x")[:-1],
             fs=fsspec.filesystem(self.protocol, **options),
             auto_mkdir=True,

src/cdf/core/project.py

@@ -48,6 +48,7 @@
 from functools import cached_property
 from pathlib import Path
 
+import duckdb
 import dynaconf
 import pydantic
 from dynaconf.utils.boxing import DynaBox
@@ -753,6 +754,25 @@ def feature_flags(self) -> FeatureFlagAdapter:
             self.ff_settings, filesystem=self.filesystem
         ).unwrap()
 
+    @cached_property
+    def duckdb(self) -> duckdb.DuckDBPyConnection:
+        """Get a handle to the project's DuckDB connection"""
+        conn = duckdb.connect(":memory:")
+        conn.install_extension("httpfs")
+        conn.install_extension("json")
+        conn.register_filesystem(self.filesystem.wrapped)
+        conn.execute("CREATE TABLE workspaces (name TEXT PRIMARY KEY, path TEXT)")
+        for workspace in self.workspaces:
+            conn.execute(
+                "INSERT INTO workspaces (name, path) VALUES (?, ?)",
+                (workspace.name, workspace.path.as_posix()),
+            )
+        return conn
+
+    def get_workspace_path(self, name: str) -> M.Result[Path, Exception]:
+        """Get the path to a workspace by name"""
+        return self.get_workspace(name).map(lambda ws: ws.path)
+
     @classmethod
     def from_path(cls, root: PathLike):
         """Load configuration data from a project root path using dynaconf.