7237750436

Committed 17 Dec 2023 09:57AM UTC coverage: 87.419% (-0.08%) from 87.497%

Build # 7237750436

Build Type

push

github

Committed by

FlorianWilhelm

Commit Message

Simplify 'show' and some refactorings

Run Details

541 of 735 branches covered (0.0%)

Branch coverage included in aggregate %.

48 of 61 new or added lines in 5 files covered. (78.69%)

2 existing lines in 2 files now uncovered.

3392 of 3764 relevant lines covered (90.12%)

4.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.23

/src/ultimate_notion/view.py

"""View representing the result of a Query."""

from __future__ import annotations

from collections.abc import Callable
from html import escape as htmlescape
from typing import TYPE_CHECKING, Any, TypeAlias, TypeVar

import numpy as np
import pandas as pd
from tabulate import tabulate

from ultimate_notion.obj_api.query import DBQueryBuilder
from ultimate_notion.objects import File, RichText
from ultimate_notion.page import Page
from ultimate_notion.text import html_img
from ultimate_notion.utils import SList, deepcopy_with_sharing, find_index, find_indices, get_repr, is_notebook

if TYPE_CHECKING:
    from ultimate_notion.database import Database

ColType: TypeAlias = str | list[str]
T = TypeVar('T')


class Col(str):
    """Column in a view, behaving like a string with some special attributes."""

    # ToDo: Implement me, in order to allow Spark-like sorting/filtering, e.g. `Col(Name) == "Florian"` in View


class View:  # noqa: PLR0904
    def __init__(self, database: Database, pages: list[Page], query: DBQueryBuilder):
        self.database = database
        self._query = query
        self._title_col = database.schema.get_title_prop().name
        self._columns = self._get_columns(self._title_col)
        self._pages: np.ndarray = np.array(pages)
        self.default_limit = 10

        self.reset()

    def reset(self) -> View:
        """Reset the view, i.e. remove filtering, index and sorting."""
        self._icon_name: str | None = None
        self._id_name: str | None = None
        self._index_name: str | None = None
        self._row_indices = np.arange(len(self._pages))
        self._col_indices = np.arange(len(self._columns))
        return self

    def clone(self) -> View:
        """Clone the current view."""
        return deepcopy_with_sharing(self, shared_attributes=['database', '_pages', '_query'])

    def _get_columns(self, title_col: str) -> np.ndarray:
        """Make sure title column is the first columns."""
        cols = list(self.database.schema.to_dict().keys())
        cols.insert(0, cols.pop(cols.index(title_col)))
        return np.array(cols)

    @property
    def columns(self) -> list[str]:
        """Columns of the database view aligned with the elements of a row."""
        cols = list(self._columns[self._col_indices])
        if self.has_icon:
            cols.insert(0, self._icon_name)
        if self.has_id:
            cols.insert(0, self._id_name)
        if self.has_index:
            cols.insert(0, self._index_name)
        return cols

    def get_page(self, idx: int) -> Page:
        """Retrieve a page by index of the view."""
        return self._pages[self._row_indices[idx]]

    def search_page(self, name: str | RichText) -> SList[Page]:
        """Retrieve a page from this view by name"""
        pages = [page for page in self.to_pages() if page.title == name]
        return SList(pages)

    def get_row(self, idx: int) -> tuple[Any, ...]:
        page = self.get_page(idx)
        row: list[Any] = []
        for col in self.columns:
            if col == self._title_col:
                row.append(page.title)
            elif col == self._id_name:
                row.append(page.id)
            elif col == self._index_name:
                row.append(idx)
            elif col == self._icon_name:
                row.append(page.icon)
            else:
                row.append(page.props[col])
        return tuple(row)

    def to_pages(self) -> list[Page]:
        """Convert the view to a simple list of pages."""
        return [self.get_page(idx) for idx in range(len(self))]

    def to_rows(self) -> list[tuple[Any, ...]]:
        return [self.get_row(idx) for idx in range(len(self))]

    def to_pandas(self) -> pd.DataFrame:
        """Convert the view to a pandas dataframe."""
        # remove index as pandas uses its own
        view = self.without_index() if self.has_index else self
        return pd.DataFrame(view.to_rows(), columns=view.columns)

    def _html_for_icon(self, rows: list[Any], cols: list[str]) -> list[Any]:
        # escape everything as we ask tabulate not to do it
        rows = [[htmlescape(elem) if isinstance(elem, str) else elem for elem in row] for row in rows]
        if (title_idx := find_index(self._icon_name, cols)) is None:
            return rows
        for idx, row in enumerate(rows):
            page = self.get_page(idx)
            icon = html_img(page.icon.url, size=1.2) if isinstance(page.icon, File) else page.icon
            row[title_idx] = icon
        return rows

    def as_table(self, tablefmt: str | None = None) -> str:
        """Return the view in a given string table format.

        Some table formats:

        - plain: no pseudographics
        - simple: Pandoc's simple table, i.e. only dashes to separate header from content
        - github: GitHub flavored Markdown
        - simple_grid: uses dashes & pipes to separate cells
        - html: standard html markup

        Find more table formats under: https://github.com/astanin/python-tabulate#table-format
        """
        rows = self.to_rows()
        cols = self.columns

        if tablefmt is None:
            tablefmt = 'html' if is_notebook() else 'simple'

        if tablefmt == 'html':
            if self.has_icon:
                rows = self._html_for_icon(rows, cols)
                html_str = str(tabulate(rows, headers=cols, tablefmt='unsafehtml'))  # str() as tabulate wraps the str
            else:
                html_str = str(tabulate(rows, headers=cols, tablefmt='html'))  # str() as tabulate wraps the str
            return html_str
        else:
            return tabulate(rows, headers=cols, tablefmt=tablefmt)

    def show(self, *, simple: bool | None = None):
        """Show the database as human-readable table."""
        if simple:
            tablefmt = 'simple'
        elif simple is None:
            tablefmt = 'html' if is_notebook() else 'simple'
        else:
            tablefmt = 'html'

        table_str = self.as_table(tablefmt=tablefmt)

        if is_notebook() and (tablefmt == 'html'):
            from IPython.display import display_html  # noqa: PLC0415

            display_html(table_str)
        else:
            print(table_str)  # noqa: T201

    def _repr_html_(self) -> str:  # noqa: PLW3201
        """Called by Jupyter Lab automatically to display this view."""
        if len(self) > self.default_limit:
            html_str = self.limit(self.default_limit).as_table(tablefmt='html')
            html_str = f'<div style="display: inline-block;text-align: center;">{html_str}&#8942;</div></div>\n'
            return html_str
        else:
            return self.as_table(tablefmt='html')

    def __repr__(self) -> str:
        return get_repr(self, desc=self.database.title)

    def __str__(self) -> str:
        return self.as_table()

    def __len__(self):
        return len(self._row_indices)

    @property
    def has_index(self) -> bool:
        return self._index_name is not None

    def with_index(self, name='index') -> View:
        """Add an index column to the view."""
        if self.has_index and name == self._index_name:
            return self

        if name in self.columns:
            msg = f"index '{name}' is already a column name"
            raise RuntimeError(msg)

        view = self.clone()
        view._index_name = name
        return view

    def without_index(self) -> View:
        """Remove index column from the view."""
        if not self.has_index:
            return self

        view = self.clone()
        view._index_name = None
        return view

    @property
    def has_icon(self) -> bool:
        return self._icon_name is not None

    def with_icon(self, name='icon') -> View:
        """Show icons in HTML output."""
        if self.has_icon and name == self._icon_name:
            return self

        view = self.clone()
        view._icon_name = name
        return view

    def without_icon(self) -> View:
        """Don't show icons in HTML output."""
        if not self.has_icon:
            return self

        view = self.clone()
        view._icon_name = None
        return view

    @property
    def has_id(self) -> bool:
        return self._id_name is not None

    def with_id(self, name: str = 'id') -> View:
        """Add an id column to the view."""
        if self.has_id and name == self._id_name:
            return self

        view = self.clone()
        view._id_name = name
        return view

    def without_id(self) -> View:
        """Remove id column from the view."""
        if not self.has_id:
            return self

        view = self.clone()
        view._id_name = None
        return view

    def head(self, num: int) -> View:
        """Keep only the first `num` elements in view."""
        view = self.clone()
        view._row_indices = view._row_indices[:num]
        return view

    def limit(self, num: int) -> View:
        """Alias for `head`"""
        return self.head(num)

    def tail(self, num: int) -> View:
        """Keep only the last `num` elements in view."""
        view = self.clone()
        view._row_indices = view._row_indices[-num:]
        return view

    def select(self, *cols: str) -> View:
        """Select columns for the view"""
        curr_cols = self._columns  # we only consider non-meta columns, e.g. no index, etc.
        if not_included := set(cols) - set(curr_cols):
            msg = f"Some columns, i.e. {', '.join(not_included)}, are not in view"
            raise RuntimeError(msg)

        view = self.clone()
        select_col_indices = find_indices(list(cols), curr_cols)
        view._col_indices = view._col_indices[select_col_indices]
        return view

    def apply(self, func: Callable[[Page], T]) -> list[T]:
        """Apply function to all pages in view.

        Args:
            func: function taking a Page as input
        """
        return [func(page) for page in self.to_pages()]

    def reverse(self) -> View:
        """Reverse the order of the rows."""
        view = self.clone()
        view._row_indices = view._row_indices[::-1]
        return view

    def sort(self):
        """Sort the view with respect to some columns."""
        # ToDo: Implement me
        raise NotImplementedError

    def filter(self):  # noqa: A003
        """Filter the view."""
        # ToDo: Implement me
        raise NotImplementedError

    def reload(self) -> View:
        """Reload all pages by re-executing the query that generated the view."""
        view = self.clone()
        view._pages = np.array(self.database._pages_from_query(query=self._query))
        return view

1	"""View representing the result of a Query."""	5✔
2
3	from __future__ import annotations	5✔
4
5	from collections.abc import Callable	5✔
6	from html import escape as htmlescape	5✔
7	from typing import TYPE_CHECKING, Any, TypeAlias, TypeVar	5✔
8
9	import numpy as np	5✔
10	import pandas as pd	5✔
11	from tabulate import tabulate	5✔
12
13	from ultimate_notion.obj_api.query import DBQueryBuilder	5✔
14	from ultimate_notion.objects import File, RichText	5✔
15	from ultimate_notion.page import Page	5✔
16	from ultimate_notion.text import html_img	5✔
17	from ultimate_notion.utils import SList, deepcopy_with_sharing, find_index, find_indices, get_repr, is_notebook	5✔
18
19	if TYPE_CHECKING:
20	from ultimate_notion.database import Database
21
22	ColType: TypeAlias = str \| list[str]	5✔
23	T = TypeVar('T')	5✔
24
25
26	class Col(str):	5✔
27	"""Column in a view, behaving like a string with some special attributes."""	5✔
28
29	# ToDo: Implement me, in order to allow Spark-like sorting/filtering, e.g. `Col(Name) == "Florian"` in View
30
31
32	class View: # noqa: PLR0904	5✔
33	def __init__(self, database: Database, pages: list[Page], query: DBQueryBuilder):	5✔
34	self.database = database	5✔
35	self._query = query	5✔
36	self._title_col = database.schema.get_title_prop().name	5✔
37	self._columns = self._get_columns(self._title_col)	5✔
38	self._pages: np.ndarray = np.array(pages)	5✔
39	self.default_limit = 10	5✔
40
41	self.reset()	5✔
42
43	def reset(self) -> View:	5✔
44	"""Reset the view, i.e. remove filtering, index and sorting."""
45	self._icon_name: str \| None = None	5✔
46	self._id_name: str \| None = None	5✔
47	self._index_name: str \| None = None	5✔
48	self._row_indices = np.arange(len(self._pages))	5✔
49	self._col_indices = np.arange(len(self._columns))	5✔
50	return self	5✔
51
52	def clone(self) -> View:	5✔
53	"""Clone the current view."""
54	return deepcopy_with_sharing(self, shared_attributes=['database', '_pages', '_query'])	5✔
55
56	def _get_columns(self, title_col: str) -> np.ndarray:	5✔
57	"""Make sure title column is the first columns."""
58	cols = list(self.database.schema.to_dict().keys())	5✔
59	cols.insert(0, cols.pop(cols.index(title_col)))	5✔
60	return np.array(cols)	5✔
61
62	@property	5✔
63	def columns(self) -> list[str]:	5✔
64	"""Columns of the database view aligned with the elements of a row."""
65	cols = list(self._columns[self._col_indices])	5✔
66	if self.has_icon:	5✔
67	cols.insert(0, self._icon_name)	×
68	if self.has_id:	5✔
69	cols.insert(0, self._id_name)	×
70	if self.has_index:	5✔
71	cols.insert(0, self._index_name)	5✔
72	return cols	5✔
73
74	def get_page(self, idx: int) -> Page:	5✔
75	"""Retrieve a page by index of the view."""
76	return self._pages[self._row_indices[idx]]	5✔
77
78	def search_page(self, name: str \| RichText) -> SList[Page]:	5✔
79	"""Retrieve a page from this view by name"""
80	pages = [page for page in self.to_pages() if page.title == name]	4✔
81	return SList(pages)	4!
82
83	def get_row(self, idx: int) -> tuple[Any, ...]:	5✔
84	page = self.get_page(idx)	5✔
85	row: list[Any] = []	5✔
86	for col in self.columns:	5✔
87	if col == self._title_col:	5✔
88	row.append(page.title)	5✔
89	elif col == self._id_name:	5✔
90	row.append(page.id)	×
91	elif col == self._index_name:	5✔
92	row.append(idx)	5✔
93	elif col == self._icon_name:	5✔
94	row.append(page.icon)	×
95	else:
96	row.append(page.props[col])	5✔
97	return tuple(row)	5✔
98
99	def to_pages(self) -> list[Page]:	5✔
100	"""Convert the view to a simple list of pages."""
101	return [self.get_page(idx) for idx in range(len(self))]	4✔
102
103	def to_rows(self) -> list[tuple[Any, ...]]:	5✔
104	return [self.get_row(idx) for idx in range(len(self))]	5✔
105
106	def to_pandas(self) -> pd.DataFrame:	5✔
107	"""Convert the view to a pandas dataframe."""
108	# remove index as pandas uses its own
109	view = self.without_index() if self.has_index else self	5✔
110	return pd.DataFrame(view.to_rows(), columns=view.columns)	5✔
111
112	def _html_for_icon(self, rows: list[Any], cols: list[str]) -> list[Any]:	5✔
113	# escape everything as we ask tabulate not to do it
114	rows = [[htmlescape(elem) if isinstance(elem, str) else elem for elem in row] for row in rows]	×
115	if (title_idx := find_index(self._icon_name, cols)) is None:	×
116	return rows	×
117	for idx, row in enumerate(rows):	×
118	page = self.get_page(idx)	×
119	icon = html_img(page.icon.url, size=1.2) if isinstance(page.icon, File) else page.icon	×
120	row[title_idx] = icon	×
121	return rows	×
122
123	def as_table(self, tablefmt: str \| None = None) -> str:	5✔
124	"""Return the view in a given string table format.
125
126	Some table formats:
127
128	- plain: no pseudographics
129	- simple: Pandoc's simple table, i.e. only dashes to separate header from content
130	- github: GitHub flavored Markdown
131	- simple_grid: uses dashes & pipes to separate cells
132	- html: standard html markup
133
134	Find more table formats under: https://github.com/astanin/python-tabulate#table-format
135	"""
136	rows = self.to_rows()	4✔
137	cols = self.columns	4✔
138
139	if tablefmt is None:	4✔
UNCOV 140	tablefmt = 'html' if is_notebook() else 'simple'	×
141
142	if tablefmt == 'html':	4✔
143	if self.has_icon:	×
144	rows = self._html_for_icon(rows, cols)	×
145	html_str = str(tabulate(rows, headers=cols, tablefmt='unsafehtml')) # str() as tabulate wraps the str	×
146	else:
147	html_str = str(tabulate(rows, headers=cols, tablefmt='html')) # str() as tabulate wraps the str	×
148	return html_str	×
149	else:
150	return tabulate(rows, headers=cols, tablefmt=tablefmt)	4✔
151
152	def show(self, *, simple: bool \| None = None):	5✔
153	"""Show the database as human-readable table."""
154	if simple:	4✔
155	tablefmt = 'simple'	4✔
156	elif simple is None:	4✔
157	tablefmt = 'html' if is_notebook() else 'simple'	4!
158	else:
NEW 159	tablefmt = 'html'	×
160
161	table_str = self.as_table(tablefmt=tablefmt)	4✔
162
163	if is_notebook() and (tablefmt == 'html'):	4✔
NEW 164	from IPython.display import display_html # noqa: PLC0415	×
165
NEW 166	display_html(table_str)	×
167	else:
168	print(table_str) # noqa: T201	4✔
169
170	def _repr_html_(self) -> str: # noqa: PLW3201	5✔
171	"""Called by Jupyter Lab automatically to display this view."""
172	if len(self) > self.default_limit:	×
173	html_str = self.limit(self.default_limit).as_table(tablefmt='html')	×
174	html_str = f'<div style="display: inline-block;text-align: center;">{html_str}⋮</div></div>\n'	×
175	return html_str	×
176	else:
177	return self.as_table(tablefmt='html')	×
178
179	def __repr__(self) -> str:
180	return get_repr(self, desc=self.database.title)
181
182	def __str__(self) -> str:	5✔
183	return self.as_table()	×
184
185	def __len__(self):	5✔
186	return len(self._row_indices)	5✔
187
188	@property	5✔
189	def has_index(self) -> bool:	5✔
190	return self._index_name is not None	5✔
191
192	def with_index(self, name='index') -> View:	5✔
193	"""Add an index column to the view."""
194	if self.has_index and name == self._index_name:	5✔
195	return self	5✔
196
197	if name in self.columns:	5✔
198	msg = f"index '{name}' is already a column name"	×
199	raise RuntimeError(msg)	×
200
201	view = self.clone()	5✔
202	view._index_name = name	5✔
203	return view	5✔
204
205	def without_index(self) -> View:	5✔
206	"""Remove index column from the view."""
207	if not self.has_index:	5✔
208	return self	5✔
209
210	view = self.clone()	5✔
211	view._index_name = None	5✔
212	return view	5✔
213
214	@property	5✔
215	def has_icon(self) -> bool:	5✔
216	return self._icon_name is not None	5✔
217
218	def with_icon(self, name='icon') -> View:	5✔
219	"""Show icons in HTML output."""
220	if self.has_icon and name == self._icon_name:	4✔
221	return self	×
222
223	view = self.clone()	4✔
224	view._icon_name = name	4✔
225	return view	4✔
226
227	def without_icon(self) -> View:	5✔
228	"""Don't show icons in HTML output."""
229	if not self.has_icon:	×
230	return self	×
231
232	view = self.clone()	×
233	view._icon_name = None	×
234	return view	×
235
236	@property	5✔
237	def has_id(self) -> bool:	5✔
238	return self._id_name is not None	5✔
239
240	def with_id(self, name: str = 'id') -> View:	5✔
241	"""Add an id column to the view."""
242	if self.has_id and name == self._id_name:	4✔
243	return self	×
244
245	view = self.clone()	4✔
246	view._id_name = name	4✔
247	return view	4✔
248
249	def without_id(self) -> View:	5✔
250	"""Remove id column from the view."""
251	if not self.has_id:	×
252	return self	×
253
254	view = self.clone()	×
255	view._id_name = None	×
256	return view	×
257
258	def head(self, num: int) -> View:	5✔
259	"""Keep only the first `num` elements in view."""
260	view = self.clone()	5✔
261	view._row_indices = view._row_indices[:num]	5✔
262	return view	5✔
263
264	def limit(self, num: int) -> View:	5✔
265	"""Alias for `head`"""
266	return self.head(num)	5✔
267
268	def tail(self, num: int) -> View:	5✔
269	"""Keep only the last `num` elements in view."""
270	view = self.clone()	×
271	view._row_indices = view._row_indices[-num:]	×
272	return view	×
273
274	def select(self, *cols: str) -> View:	5✔
275	"""Select columns for the view"""
276	curr_cols = self._columns # we only consider non-meta columns, e.g. no index, etc.	5✔
277	if not_included := set(cols) - set(curr_cols):	5✔
278	msg = f"Some columns, i.e. {', '.join(not_included)}, are not in view"	5✔
279	raise RuntimeError(msg)	5✔
280
281	view = self.clone()	5✔
282	select_col_indices = find_indices(list(cols), curr_cols)	5✔
283	view._col_indices = view._col_indices[select_col_indices]	5✔
284	return view	5✔
285
286	def apply(self, func: Callable[[Page], T]) -> list[T]:	5✔
287	"""Apply function to all pages in view.
288
289	Args:
290	func: function taking a Page as input
291	"""
292	return [func(page) for page in self.to_pages()]	×
293
294	def reverse(self) -> View:	5✔
295	"""Reverse the order of the rows."""
296	view = self.clone()	5✔
297	view._row_indices = view._row_indices[::-1]	5✔
298	return view	5✔
299
300	def sort(self):	5✔
301	"""Sort the view with respect to some columns."""
302	# ToDo: Implement me
303	raise NotImplementedError
304
305	def filter(self): # noqa: A003	5✔
306	"""Filter the view."""
307	# ToDo: Implement me
308	raise NotImplementedError
309
310	def reload(self) -> View:	5✔
311	"""Reload all pages by re-executing the query that generated the view."""
312	view = self.clone()	×
313	view._pages = np.array(self.database._pages_from_query(query=self._query))	×
314	return view	×

ultimate-notion / ultimate-notion / 7237750436

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous