Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/jsoniq/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from jsoniq.errors import RumbleError
from jsoniq.session import RumbleSession

__all__ = ["RumbleSession"]
__all__ = ["RumbleSession", "RumbleError"]
48 changes: 48 additions & 0 deletions src/jsoniq/errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
"""Notebook-friendly Java errors that retain normal exception semantics."""

from contextlib import contextmanager
from functools import wraps

from py4j.protocol import Py4JJavaError


class RumbleError(Py4JJavaError):
"""A Java query error displayed without a traceback by IPython.

Remains a Py4JJavaError for compatibility with existing exception handlers.
Ordinary Python execution still reports a traceback.
"""

def __init__(self, original):
super().__init__(original.args[0], original.java_exception)
self.message = original.java_exception.getMessage()

def __str__(self):
return self.message or self.java_exception.toString()

def _render_traceback_(self):
return str(self).splitlines()


@contextmanager
def notebook_error_context(session):
"""Apply the same error policy to an entire query execution block."""
try:
yield
except RumbleError:
raise
except Py4JJavaError as error:
if session.getRumbleConf().getBoolean("debug.showErrorInfo"):
raise
raise RumbleError(error) from error


def notebook_errors(method):
"""Translate Java errors unless full debugging output was requested."""
@wraps(method)
def wrapped(self, *args, **kwargs):
session = self.__dict__.get("_rumblesession", self)
with notebook_error_context(session):
return method(self, *args, **kwargs)

return wrapped
18 changes: 17 additions & 1 deletion src/jsoniq/sequence.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from pyspark.sql import DataFrame
import json
import sys
from .errors import notebook_errors

class SequenceOfItems:
schema_str = """
Expand Down Expand Up @@ -32,40 +33,55 @@ def __init__(self, sequence, rumblesession):
self._sparksession = rumblesession._sparksession
self._sparkcontext = self._sparksession.sparkContext

@notebook_errors
def items(self):
return self.getAsList()

@notebook_errors
def take(self, n):
self._rumblesession.lastResult = tuple(self.getFirstItemsAsList(n))
return self._rumblesession.lastResult

@notebook_errors
def first(self):
self._rumblesession.lastResult = tuple(self.getFirstItemsAsList(self._rumblesession.getRumbleConf().getInt("runtime.resultsSizeCap")))
return self._rumblesession.lastResult

@notebook_errors
def json(self):
self._rumblesession.lastResult = tuple([json.loads(l.serializeAsJSON()) for l in self._jsequence.getAsList()])
return self._rumblesession.lastResult

@notebook_errors
def rdd(self):
rdd = self._jsequence.getAsPickledStringRDD()
rdd = RDD(rdd, self._sparkcontext)
self._rumblesession.lastResult = rdd.map(lambda l: json.loads(l))
return self._rumblesession.lastResult

@notebook_errors
def df(self):
self._rumblesession.lastResult = DataFrame(self._jsequence.getAsDataFrame(), self._sparksession)
return self._rumblesession.lastResult

@notebook_errors
def pdf(self):
self._rumblesession.lastResult = self.df().toPandas()
return self._rumblesession.lastResult

@notebook_errors
def count(self):
return self._jsequence.count()

@notebook_errors
def nextJSON(self):
return self._jsequence.next().serializeAsJSON()

def __getattr__(self, item):
return getattr(self._jsequence, item)
value = getattr(self._jsequence, item)
if callable(value):
@notebook_errors
def call(sequence, *args, **kwargs):
return value(*args, **kwargs)
return call.__get__(self, type(self))
return value
3 changes: 3 additions & 0 deletions src/jsoniq/session.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from pyspark.sql import SparkSession
from py4j.java_collections import JavaList
from .errors import notebook_errors
from .sequence import SequenceOfItems
from .configuration import RumbleConfiguration
import sys
Expand Down Expand Up @@ -333,9 +334,11 @@ def bindDataFrameAsVariable(self, name: str, df):
self._bindings[name] = ("bindDataFrame", df._jdf)
return self;

@notebook_errors
def jsoniq(self, str, **kwargs):
return self._run_query(str, self._jrumblesession, kwargs)

@notebook_errors
def xquery(self, str, **kwargs):
"""Run a query with XQuery 3.1 as the default language."""
builder = self._jrumblesession.getConfiguration().toBuilder()
Expand Down
138 changes: 12 additions & 126 deletions src/jsoniqmagic/magic.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
)
import time, json
from jsoniq.session import RumbleSession
from py4j.protocol import Py4JJavaError
from jsoniq.errors import notebook_error_context

@magics_class
class JSONiqMagic(Magics):
Expand Down Expand Up @@ -39,153 +39,39 @@ def run(self, line, cell=None, timed=False, language="jsoniq"):
):
args.serialize = True
start = time.time()
try:
rumble = RumbleSession.builder.getOrCreate();
response = getattr(rumble, language)(data);
except Py4JJavaError as e:
print(e.java_exception.getMessage())
return
except Exception as e:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
print("Error message:")
print(e.args[0])
return
except:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
return
rumble = RumbleSession.builder.getOrCreate();
with notebook_error_context(rumble):
return self._execute(rumble, language, data, args, start)

def _execute(self, rumble, language, data, args, start):
response = getattr(rumble, language)(data);

if args.serialize:
try:
serialized = response.serialize()
except Py4JJavaError as e:
if rumble.getRumbleConf().getBoolean("debug.showErrorInfo"):
raise
print(e.java_exception.getMessage())
return
except Exception as e:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
print("Error message:")
print(e.args[0])
return
except:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
return
serialized = response.serialize()
print(serialized)
if args.timed:
print("Response time: %s ms" % (time.time() - start))
return

schema_str = """
No DataFrame available as no schema was detected. If you still believe the output is structured enough, you could add a schema and validate expression explicitly to your query.

This is an example of how you can simply define a schema and wrap your query in a validate expression:

declare type mytype as {
"product" : "string",
"store-number" : "int",
"quantity" : "decimal"
};
validate type mytype* {
for $product in json-lines("http://rumbledb.org/samples/products-small.json", 10)
where $product.quantity ge 995
return $product
}
"""

if(args.pyspark_data_frame):
try:
df = response.df();
except Py4JJavaError as e:
if rumble.getRumbleConf().getBoolean("debug.showErrorInfo") :
raise e;
else:
print(e.java_exception.getMessage())
return
except Exception as e:
if rumble.getRumbleConf().getBoolean("debug.showErrorInfo") :
raise e;
else:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
print("Error message:")
print(e.args[0])
return
except:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
return
df = response.df();
if df is not None:
df.show()

if (args.pandas_data_frame):
try:
pdf = response.pdf()
except Py4JJavaError as e:
if rumble.getRumbleConf().getBoolean("debug.showErrorInfo") :
raise e;
else:
print(e.java_exception.getMessage())
return
except Exception as e:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
print("Error message:")
print(e.args[0])
return
except:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
return
pdf = response.pdf()
if pdf is not None:
return pdf

if (args.apply_updates):
if ("PUL" in response.availableOutputs()):
try:
response.applyPUL()
except Py4JJavaError as e:
if rumble.getRumbleConf().getBoolean("debug.showErrorInfo") :
raise e;
else:
print(e.java_exception.getMessage())
return
except Exception as e:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
print("Error message:")
print(e.args[0])
return
except:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
return
response.applyPUL()
print("Updates applied successfully.")
else:
print("No Pending Update List (PUL) available to apply.")

if (args.json or (not args.pandas_data_frame and not args.pyspark_data_frame)):
try:
capplusone = response.take(rumble.getRumbleConf().getInt("runtime.resultsSizeCap") + 1)
except Py4JJavaError as e:
if rumble.getRumbleConf().getBoolean("debug.showErrorInfo") :
raise e;
else:
print(e.java_exception.getMessage())
return
except Exception as e:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
print("Error message:")
print(e.args[0])
return
except:
print("Query unsuccessful.")
print("Usual reasons: firewall, misconfigured proxy.")
return
capplusone = response.take(rumble.getRumbleConf().getInt("runtime.resultsSizeCap") + 1)
if len(capplusone) > rumble.getRumbleConf().getInt("runtime.resultsSizeCap"):
count = response.count()
print("The query output %s items, which is too many to display. Displaying the first %s items:" % (count, rumble.getRumbleConf().getInt("runtime.resultsSizeCap")))
Expand Down
7 changes: 3 additions & 4 deletions tests/test_magic_serialization.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,14 +27,13 @@ def test_magic_prints_serialized_sequence_directly(option, serialized, capsys):
assert capsys.readouterr().out == serialized + "\n"


def test_magic_reports_serialization_errors(capsys):
def test_magic_preserves_python_serialization_errors():
session = Mock()
session.jsoniq.return_value.serialize.side_effect = RuntimeError("Serialization failed")

with patch.object(RumbleSession.Builder, "getOrCreate", return_value=session):
JSONiqMagic().jsoniq("-s", "1 to 3")

assert "Serialization failed" in capsys.readouterr().out
with pytest.raises(RuntimeError, match="Serialization failed"):
JSONiqMagic().jsoniq("-s", "1 to 3")


def test_magic_serialization_preserves_timing(capsys):
Expand Down
Loading
Loading