Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(datasets): Added the Experimental PolarsDatabaseDataset #990

Open
wants to merge 13 commits into
base: main
Choose a base branch
from
Open
11 changes: 11 additions & 0 deletions kedro-datasets/kedro_datasets_experimental/polars/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
"""``AbstractDataset`` implementation to load/save to databases using the Polars library."""

from typing import Any

import lazy_loader as lazy

PolarsDatabaseDataset: Any

__getattr__, __dir__, __all__ = lazy.attach(
__name__, submod_attrs={"polars_database_dataset": ["PolarsDatabaseDataset"]}
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
import copy
from pathlib import PurePosixPath
from typing import Any, NoReturn

import polars as pl

from kedro_datasets.pandas.sql_dataset import SQLQueryDataset, get_filepath_str


class PolarsDatabaseDataset(SQLQueryDataset):

def __init__( # noqa: PLR0913
self,
MinuraPunchihewa marked this conversation as resolved.
Show resolved Hide resolved
sql: str | None = None,
credentials: dict[str, Any] | None = None,
load_args: dict[str, Any] | None = None,
fs_args: dict[str, Any] | None = None,
filepath: str | None = None,
metadata: dict[str, Any] | None = None,
) -> None:
"""Creates a new ``PolarsDatabaseDataset``."""
super().__init__(
sql=sql,
credentials=credentials,
load_args=load_args,
fs_args=fs_args,
filepath=filepath,
metadata=metadata,
)

def load(self) -> pl.DataFrame:
load_args = copy.deepcopy(self._load_args)

if self._filepath:
load_path = get_filepath_str(PurePosixPath(self._filepath), self._protocol)
with self._fs.open(load_path, mode="r") as fs_file:
query = fs_file.read()
else:
query = load_args.pop("sql")

return pl.read_database(
query=query,
connection=self._connection_str,
**load_args
Comment on lines +254 to +267
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you add some docs or checking to reflect this logic

  1. If filepath exist use it
  2. Otherwise use sql
  3. If both are defined, maybe error out or at least log which one is being used?

Ideally I would re-order the argument as well, since the dataset put sql as the first argument but actually filepath have higher priority which feels counter-intuitive.

)

def save(self, data: None) -> NoReturn:
pass
MinuraPunchihewa marked this conversation as resolved.
Show resolved Hide resolved
Loading