Родитель
c5773c5b4d
Коммит
5fa025fb02
|
@ -4,7 +4,6 @@ from datetime import timedelta
|
||||||
import re
|
import re
|
||||||
from dateutil import parser
|
from dateutil import parser
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
import six
|
|
||||||
|
|
||||||
# Regex for TimeSpan
|
# Regex for TimeSpan
|
||||||
_TIMESPAN_PATTERN = re.compile(r"(-?)((?P<d>[0-9]*).)?(?P<h>[0-9]{2}):(?P<m>[0-9]{2}):(?P<s>[0-9]{2}(\.[0-9]+)?$)")
|
_TIMESPAN_PATTERN = re.compile(r"(-?)((?P<d>[0-9]*).)?(?P<h>[0-9]{2}):(?P<m>[0-9]{2}):(?P<s>[0-9]{2}(\.[0-9]+)?$)")
|
||||||
|
@ -12,14 +11,14 @@ _TIMESPAN_PATTERN = re.compile(r"(-?)((?P<d>[0-9]*).)?(?P<h>[0-9]{2}):(?P<m>[0-9
|
||||||
|
|
||||||
def to_datetime(value):
|
def to_datetime(value):
|
||||||
"""Converts a string to a datetime."""
|
"""Converts a string to a datetime."""
|
||||||
if isinstance(value, six.integer_types):
|
if isinstance(value, int):
|
||||||
return parser.parse(value)
|
return parser.parse(value)
|
||||||
return parser.isoparse(value)
|
return parser.isoparse(value)
|
||||||
|
|
||||||
|
|
||||||
def to_timedelta(value):
|
def to_timedelta(value):
|
||||||
"""Converts a string to a timedelta."""
|
"""Converts a string to a timedelta."""
|
||||||
if isinstance(value, (six.integer_types, float)):
|
if isinstance(value, (int, float)):
|
||||||
return timedelta(microseconds=(float(value) / 10))
|
return timedelta(microseconds=(float(value) / 10))
|
||||||
match = _TIMESPAN_PATTERN.match(value)
|
match = _TIMESPAN_PATTERN.match(value)
|
||||||
if match:
|
if match:
|
||||||
|
@ -27,6 +26,6 @@ def to_timedelta(value):
|
||||||
factor = -1
|
factor = -1
|
||||||
else:
|
else:
|
||||||
factor = 1
|
factor = 1
|
||||||
return factor * timedelta(days=int(match.group("d") or 0), hours=int(match.group("h")), minutes=int(match.group("m")), seconds=float(match.group("s")),)
|
return factor * timedelta(days=int(match.group("d") or 0), hours=int(match.group("h")), minutes=int(match.group("m")), seconds=float(match.group("s")))
|
||||||
else:
|
else:
|
||||||
raise ValueError("Timespan value '{}' cannot be decoded".format(value))
|
raise ValueError("Timespan value '{}' cannot be decoded".format(value))
|
||||||
|
|
|
@ -4,7 +4,6 @@ import json
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
import six
|
|
||||||
from . import _converters
|
from . import _converters
|
||||||
from .exceptions import KustoServiceError
|
from .exceptions import KustoServiceError
|
||||||
|
|
||||||
|
@ -76,7 +75,7 @@ class KustoResultRow(object):
|
||||||
yield self[i]
|
yield self[i]
|
||||||
|
|
||||||
def __getitem__(self, key):
|
def __getitem__(self, key):
|
||||||
if isinstance(key, six.integer_types):
|
if isinstance(key, int):
|
||||||
return self._value_by_index[key]
|
return self._value_by_index[key]
|
||||||
return self._value_by_name[key]
|
return self._value_by_name[key]
|
||||||
|
|
||||||
|
|
|
@ -5,13 +5,10 @@ import json
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
from abc import ABCMeta, abstractmethod
|
from abc import ABCMeta, abstractmethod
|
||||||
|
|
||||||
import six
|
|
||||||
|
|
||||||
from ._models import KustoResultColumn, KustoResultRow, KustoResultTable, WellKnownDataSet
|
from ._models import KustoResultColumn, KustoResultRow, KustoResultTable, WellKnownDataSet
|
||||||
|
|
||||||
|
|
||||||
@six.add_metaclass(ABCMeta)
|
class KustoResponseDataSet(metaclass=ABCMeta):
|
||||||
class KustoResponseDataSet:
|
|
||||||
"""Represents the parsed data set carried by the response to a Kusto request."""
|
"""Represents the parsed data set carried by the response to a Kusto request."""
|
||||||
|
|
||||||
def __init__(self, json_response):
|
def __init__(self, json_response):
|
||||||
|
@ -78,7 +75,7 @@ class KustoResponseDataSet:
|
||||||
return iter(self.tables)
|
return iter(self.tables)
|
||||||
|
|
||||||
def __getitem__(self, key):
|
def __getitem__(self, key):
|
||||||
if isinstance(key, six.integer_types):
|
if isinstance(key, int):
|
||||||
return self.tables[key]
|
return self.tables[key]
|
||||||
try:
|
try:
|
||||||
return self.tables[self.tables_names.index(key)]
|
return self.tables[self.tables_names.index(key)]
|
||||||
|
|
|
@ -1,5 +1,4 @@
|
||||||
"""Kusto helper functions"""
|
"""Kusto helper functions"""
|
||||||
import six
|
|
||||||
|
|
||||||
|
|
||||||
def to_pandas_datetime(raw_value, *args):
|
def to_pandas_datetime(raw_value, *args):
|
||||||
|
@ -11,11 +10,11 @@ def to_pandas_datetime(raw_value, *args):
|
||||||
def to_pandas_timedelta(raw_value, timedelta_value):
|
def to_pandas_timedelta(raw_value, timedelta_value):
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
|
|
||||||
if isinstance(raw_value, (six.integer_types, float)):
|
if isinstance(raw_value, (int, float)):
|
||||||
# https://docs.microsoft.com/en-us/dotnet/api/system.datetime.ticks
|
# https://docs.microsoft.com/en-us/dotnet/api/system.datetime.ticks
|
||||||
# kusto saves up to ticks, 1 tick == 100 nanoseconds
|
# kusto saves up to ticks, 1 tick == 100 nanoseconds
|
||||||
return pd.Timedelta(raw_value * 100, unit="ns")
|
return pd.Timedelta(raw_value * 100, unit="ns")
|
||||||
if isinstance(raw_value, six.string_types):
|
if isinstance(raw_value, str):
|
||||||
fraction = raw_value.split(".")[-1]
|
fraction = raw_value.split(".")[-1]
|
||||||
if fraction.isdigit():
|
if fraction.isdigit():
|
||||||
whole_part = int(timedelta_value.total_seconds())
|
whole_part = int(timedelta_value.total_seconds())
|
||||||
|
|
|
@ -74,13 +74,7 @@ class KustoConnectionStringBuilder(object):
|
||||||
|
|
||||||
def is_secret(self):
|
def is_secret(self):
|
||||||
"""States for each property if it contains secret"""
|
"""States for each property if it contains secret"""
|
||||||
return self in [
|
return self in [self.password, self.application_key, self.application_certificate, self.application_token, self.user_token]
|
||||||
self.password,
|
|
||||||
self.application_key,
|
|
||||||
self.application_certificate,
|
|
||||||
self.application_token,
|
|
||||||
self.user_token,
|
|
||||||
]
|
|
||||||
|
|
||||||
def is_str_type(self):
|
def is_str_type(self):
|
||||||
"""States whether a word is of type str or not."""
|
"""States whether a word is of type str or not."""
|
||||||
|
@ -102,10 +96,7 @@ class KustoConnectionStringBuilder(object):
|
||||||
|
|
||||||
def is_bool_type(self):
|
def is_bool_type(self):
|
||||||
"""States whether a word is of type bool or not."""
|
"""States whether a word is of type bool or not."""
|
||||||
return self in [
|
return self in [self.aad_federated_security, self.msi_auth]
|
||||||
self.aad_federated_security,
|
|
||||||
self.msi_auth,
|
|
||||||
]
|
|
||||||
|
|
||||||
def __init__(self, connection_string):
|
def __init__(self, connection_string):
|
||||||
"""Creates new KustoConnectionStringBuilder.
|
"""Creates new KustoConnectionStringBuilder.
|
||||||
|
@ -437,11 +428,7 @@ class KustoClient(object):
|
||||||
self._query_endpoint = "{0}/v2/rest/query".format(kusto_cluster)
|
self._query_endpoint = "{0}/v2/rest/query".format(kusto_cluster)
|
||||||
self._streaming_ingest_endpoint = "{0}/v1/rest/ingest/".format(kusto_cluster)
|
self._streaming_ingest_endpoint = "{0}/v1/rest/ingest/".format(kusto_cluster)
|
||||||
self._auth_provider = _AadHelper(kcsb) if kcsb.aad_federated_security else None
|
self._auth_provider = _AadHelper(kcsb) if kcsb.aad_federated_security else None
|
||||||
self._request_headers = {
|
self._request_headers = {"Accept": "application/json", "Accept-Encoding": "gzip,deflate", "x-ms-client-version": "Kusto.Python.Client:" + VERSION}
|
||||||
"Accept": "application/json",
|
|
||||||
"Accept-Encoding": "gzip,deflate",
|
|
||||||
"x-ms-client-version": "Kusto.Python.Client:" + VERSION,
|
|
||||||
}
|
|
||||||
|
|
||||||
def execute(self, database, query, properties=None):
|
def execute(self, database, query, properties=None):
|
||||||
"""Executes a query or management command.
|
"""Executes a query or management command.
|
||||||
|
@ -533,7 +520,7 @@ class KustoClient(object):
|
||||||
|
|
||||||
if payload:
|
if payload:
|
||||||
raise KustoServiceError(
|
raise KustoServiceError(
|
||||||
"An error occurred while trying to ingest: Status: {0.status_code}, Reason: {0.reason}, Text: {0.text}".format(response), response,
|
"An error occurred while trying to ingest: Status: {0.status_code}, Reason: {0.reason}, Text: {0.text}".format(response), response
|
||||||
)
|
)
|
||||||
|
|
||||||
raise KustoServiceError([response.json()], response)
|
raise KustoServiceError([response.json()], response)
|
||||||
|
|
|
@ -3,7 +3,7 @@ import os
|
||||||
from enum import Enum, unique
|
from enum import Enum, unique
|
||||||
from datetime import timedelta, datetime
|
from datetime import timedelta, datetime
|
||||||
import webbrowser
|
import webbrowser
|
||||||
from six.moves.urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
import dateutil.parser
|
import dateutil.parser
|
||||||
|
|
||||||
from adal import AuthenticationContext, AdalError
|
from adal import AuthenticationContext, AdalError
|
||||||
|
|
|
@ -44,6 +44,6 @@ setup(
|
||||||
namespace_packages=["azure"],
|
namespace_packages=["azure"],
|
||||||
keywords="kusto wrapper client library",
|
keywords="kusto wrapper client library",
|
||||||
packages=find_packages(exclude=["azure", "tests"]),
|
packages=find_packages(exclude=["azure", "tests"]),
|
||||||
install_requires=["adal>=1.0.0", "python-dateutil>=2.8.0", "requests>=2.13.0", "six>=1.10.0", "msrestazure>=0.4.14",],
|
install_requires=["adal>=1.0.0", "python-dateutil>=2.8.0", "requests>=2.13.0", "msrestazure>=0.4.14"],
|
||||||
extras_require={"pandas": ["pandas==0.24.1"], ":python_version<'3.0'": ["azure-nspkg"]},
|
extras_require={"pandas": ["pandas==0.24.1"], ":python_version<'3.0'": ["azure-nspkg"]},
|
||||||
)
|
)
|
||||||
|
|
|
@ -11,26 +11,26 @@ class ConverterTests(unittest.TestCase):
|
||||||
def test_to_timestamp(self):
|
def test_to_timestamp(self):
|
||||||
"""Happy path to test converter from TimeSpan to timedelta."""
|
"""Happy path to test converter from TimeSpan to timedelta."""
|
||||||
# Test hours, minutes and seconds
|
# Test hours, minutes and seconds
|
||||||
self.assertEqual(to_timedelta("00:00:00"), timedelta(seconds=0))
|
assert to_timedelta("00:00:00") == timedelta(seconds=0)
|
||||||
self.assertEqual(to_timedelta("00:00:03"), timedelta(seconds=3))
|
assert to_timedelta("00:00:03") == timedelta(seconds=3)
|
||||||
self.assertEqual(to_timedelta("00:04:03"), timedelta(minutes=4, seconds=3))
|
assert to_timedelta("00:04:03") == timedelta(minutes=4, seconds=3)
|
||||||
self.assertEqual(to_timedelta("02:04:03"), timedelta(hours=2, minutes=4, seconds=3))
|
assert to_timedelta("02:04:03") == timedelta(hours=2, minutes=4, seconds=3)
|
||||||
# Test milliseconds
|
# Test milliseconds
|
||||||
self.assertEqual(to_timedelta("00:00:00.099"), timedelta(milliseconds=99))
|
assert to_timedelta("00:00:00.099") == timedelta(milliseconds=99)
|
||||||
self.assertEqual(to_timedelta("02:04:03.0123"), timedelta(hours=2, minutes=4, seconds=3, microseconds=12300))
|
assert to_timedelta("02:04:03.0123") == timedelta(hours=2, minutes=4, seconds=3, microseconds=12300)
|
||||||
# Test days
|
# Test days
|
||||||
self.assertEqual(to_timedelta("01.00:00:00"), timedelta(days=1))
|
assert to_timedelta("01.00:00:00") == timedelta(days=1)
|
||||||
self.assertEqual(to_timedelta("02.04:05:07"), timedelta(days=2, hours=4, minutes=5, seconds=7))
|
assert to_timedelta("02.04:05:07") == timedelta(days=2, hours=4, minutes=5, seconds=7)
|
||||||
# Test negative
|
# Test negative
|
||||||
self.assertEqual(to_timedelta("-01.00:00:00"), -timedelta(days=1))
|
assert to_timedelta("-01.00:00:00") == -timedelta(days=1)
|
||||||
self.assertEqual(to_timedelta("-02.04:05:07"), -timedelta(days=2, hours=4, minutes=5, seconds=7))
|
assert to_timedelta("-02.04:05:07") == -timedelta(days=2, hours=4, minutes=5, seconds=7)
|
||||||
# Test all together
|
# Test all together
|
||||||
self.assertEqual(to_timedelta("00.00:00:00.000"), timedelta(seconds=0))
|
assert to_timedelta("00.00:00:00.000") == timedelta(seconds=0)
|
||||||
self.assertEqual(to_timedelta("02.04:05:07.789"), timedelta(days=2, hours=4, minutes=5, seconds=7, milliseconds=789))
|
assert to_timedelta("02.04:05:07.789") == timedelta(days=2, hours=4, minutes=5, seconds=7, milliseconds=789)
|
||||||
self.assertEqual(to_timedelta("03.00:00:00.111"), timedelta(days=3, milliseconds=111))
|
assert to_timedelta("03.00:00:00.111") == timedelta(days=3, milliseconds=111)
|
||||||
# Test from Ticks
|
# Test from Ticks
|
||||||
self.assertEqual(to_timedelta(-80080008), timedelta(microseconds=-8008001))
|
assert to_timedelta(-80080008) == timedelta(microseconds=-8008001)
|
||||||
self.assertEqual(to_timedelta(10010001), timedelta(microseconds=1001000))
|
assert to_timedelta(10010001) == timedelta(microseconds=1001000)
|
||||||
|
|
||||||
def test_to_timestamp_fail(self):
|
def test_to_timestamp_fail(self):
|
||||||
"""
|
"""
|
||||||
|
@ -45,7 +45,7 @@ class ConverterTests(unittest.TestCase):
|
||||||
|
|
||||||
def test_to_datetime(self):
|
def test_to_datetime(self):
|
||||||
""" Tests datetime read by KustoResultIter """
|
""" Tests datetime read by KustoResultIter """
|
||||||
self.assertIsNotNone(to_datetime("2016-06-07T16:00:00Z"))
|
assert to_datetime("2016-06-07T16:00:00Z") is not None
|
||||||
|
|
||||||
def test_to_datetime_fail(self):
|
def test_to_datetime_fail(self):
|
||||||
""" Tests that invalid strings fails to convert to datetime """
|
""" Tests that invalid strings fails to convert to datetime """
|
||||||
|
|
|
@ -3,7 +3,6 @@
|
||||||
import json
|
import json
|
||||||
import unittest
|
import unittest
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from six import text_type
|
|
||||||
from dateutil.tz.tz import tzutc
|
from dateutil.tz.tz import tzutc
|
||||||
|
|
||||||
from azure.kusto.data._response import KustoResponseDataSetV2
|
from azure.kusto.data._response import KustoResponseDataSetV2
|
||||||
|
@ -177,39 +176,31 @@ class FunctionalTests(unittest.TestCase):
|
||||||
"""Tests on happy path, validating response and iterations over it."""
|
"""Tests on happy path, validating response and iterations over it."""
|
||||||
response = KustoResponseDataSetV2(json.loads(RESPONSE_TEXT))
|
response = KustoResponseDataSetV2(json.loads(RESPONSE_TEXT))
|
||||||
# Test that basic iteration works
|
# Test that basic iteration works
|
||||||
self.assertEqual(len(response), 3)
|
assert len(response) == 3
|
||||||
self.assertEqual(len(list(response.primary_results[0])), 3)
|
assert len(list(response.primary_results[0])) == 3
|
||||||
table = list(response.tables[0])
|
table = list(response.tables[0])
|
||||||
self.assertEqual(1, len(table))
|
assert 1 == len(table)
|
||||||
|
|
||||||
expected_table = [
|
expected_table = [
|
||||||
[datetime(2016, 6, 6, 15, 35, tzinfo=tzutc()), "foo", 101, 3.14, False, timedelta(days=4, hours=1, minutes=2, seconds=3, milliseconds=567),],
|
[datetime(2016, 6, 6, 15, 35, tzinfo=tzutc()), "foo", 101, 3.14, False, timedelta(days=4, hours=1, minutes=2, seconds=3, milliseconds=567)],
|
||||||
[datetime(2016, 6, 7, 16, tzinfo=tzutc()), "bar", 555, 2.71, True, timedelta()],
|
[datetime(2016, 6, 7, 16, tzinfo=tzutc()), "bar", 555, 2.71, True, timedelta()],
|
||||||
[None, text_type(""), None, None, None, None],
|
[None, str(""), None, None, None, None],
|
||||||
]
|
]
|
||||||
|
|
||||||
|
columns = ["Timestamp", "Name", "Altitude", "Temperature", "IsFlying", "TimeFlying"]
|
||||||
|
|
||||||
# Test access by index and by column name
|
# Test access by index and by column name
|
||||||
primary_table = response.primary_results[0]
|
primary_table = response.primary_results[0]
|
||||||
for row in primary_table:
|
for row in primary_table:
|
||||||
self.assertEqual(row[0], row["Timestamp"])
|
|
||||||
self.assertEqual(row[1], row["Name"])
|
|
||||||
self.assertEqual(row[2], row["Altitude"])
|
|
||||||
self.assertEqual(row[3], row["Temperature"])
|
|
||||||
self.assertEqual(row[4], row["IsFlying"])
|
|
||||||
self.assertEqual(row[5], row["TimeFlying"])
|
|
||||||
|
|
||||||
# Test all types
|
# Test all types
|
||||||
self.assertEqual(type(row[0]), datetime if row[0] else type(None))
|
for i, t in enumerate([datetime, str, int, float, bool, timedelta]):
|
||||||
self.assertEqual(type(row[1]), text_type)
|
assert row[i] == row[columns[i]]
|
||||||
self.assertEqual(type(row[2]), int if row[2] else type(None))
|
assert row[i] is None or isinstance(row[i], t)
|
||||||
self.assertEqual(type(row[3]), float if row[3] else type(None))
|
|
||||||
self.assertEqual(type(row[4]), bool if row[4] is not None else type(None))
|
|
||||||
self.assertEqual(type(row[5]), timedelta if row[5] is not None else type(None))
|
|
||||||
|
|
||||||
for row_index, row in enumerate(primary_table):
|
for row_index, row in enumerate(primary_table):
|
||||||
expected_row = expected_table[row_index]
|
expected_row = expected_table[row_index]
|
||||||
for col_index, value in enumerate(row):
|
for col_index, value in enumerate(row):
|
||||||
self.assertEqual(value, expected_row[col_index])
|
assert value == expected_row[col_index]
|
||||||
|
|
||||||
def test_invalid_table(self):
|
def test_invalid_table(self):
|
||||||
"""Tests calling of table with index that doesn't exists."""
|
"""Tests calling of table with index that doesn't exists."""
|
||||||
|
@ -227,11 +218,11 @@ class FunctionalTests(unittest.TestCase):
|
||||||
def test_iterating_after_end(self):
|
def test_iterating_after_end(self):
|
||||||
"""Tests StopIteration is raised when the response ends."""
|
"""Tests StopIteration is raised when the response ends."""
|
||||||
response = KustoResponseDataSetV2(json.loads(RESPONSE_TEXT))
|
response = KustoResponseDataSetV2(json.loads(RESPONSE_TEXT))
|
||||||
self.assertEqual(sum(1 for _ in response.primary_results[0]), 3)
|
assert sum(1 for _ in response.primary_results[0]) == 3
|
||||||
|
|
||||||
def test_row_equality(self):
|
def test_row_equality(self):
|
||||||
"""Tests the rows are idempotent."""
|
"""Tests the rows are idempotent."""
|
||||||
response = KustoResponseDataSetV2(json.loads(RESPONSE_TEXT))
|
response = KustoResponseDataSetV2(json.loads(RESPONSE_TEXT))
|
||||||
table = response.primary_results[0]
|
table = response.primary_results[0]
|
||||||
for row_index, row in enumerate(table):
|
for row_index, row in enumerate(table):
|
||||||
self.assertEqual(table[row_index], row)
|
assert table[row_index] == row
|
||||||
|
|
|
@ -5,7 +5,6 @@ import json
|
||||||
import unittest
|
import unittest
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
import pytest
|
import pytest
|
||||||
from six import text_type
|
|
||||||
from mock import patch
|
from mock import patch
|
||||||
from dateutil.tz import UTC
|
from dateutil.tz import UTC
|
||||||
|
|
||||||
|
@ -31,7 +30,7 @@ def mocked_requests_post(*args, **kwargs):
|
||||||
|
|
||||||
def __init__(self, json_data, status_code):
|
def __init__(self, json_data, status_code):
|
||||||
self.json_data = json_data
|
self.json_data = json_data
|
||||||
self.text = text_type(json_data)
|
self.text = str(json_data)
|
||||||
self.status_code = status_code
|
self.status_code = status_code
|
||||||
self.headers = None
|
self.headers = None
|
||||||
|
|
||||||
|
@ -70,19 +69,7 @@ def mocked_requests_post(*args, **kwargs):
|
||||||
return MockResponse(None, 404)
|
return MockResponse(None, 404)
|
||||||
|
|
||||||
|
|
||||||
DIGIT_WORDS = [
|
DIGIT_WORDS = [str("Zero"), str("One"), str("Two"), str("Three"), str("Four"), str("Five"), str("Six"), str("Seven"), str("Eight"), str("Nine"), str("ten")]
|
||||||
text_type("Zero"),
|
|
||||||
text_type("One"),
|
|
||||||
text_type("Two"),
|
|
||||||
text_type("Three"),
|
|
||||||
text_type("Four"),
|
|
||||||
text_type("Five"),
|
|
||||||
text_type("Six"),
|
|
||||||
text_type("Seven"),
|
|
||||||
text_type("Eight"),
|
|
||||||
text_type("Nine"),
|
|
||||||
text_type("ten"),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
class KustoClientTests(unittest.TestCase):
|
class KustoClientTests(unittest.TestCase):
|
||||||
|
@ -95,7 +82,7 @@ class KustoClientTests(unittest.TestCase):
|
||||||
response = client.execute_query("PythonTest", "Deft")
|
response = client.execute_query("PythonTest", "Deft")
|
||||||
expected = {
|
expected = {
|
||||||
"rownumber": None,
|
"rownumber": None,
|
||||||
"rowguid": text_type(""),
|
"rowguid": str(""),
|
||||||
"xdouble": None,
|
"xdouble": None,
|
||||||
"xfloat": None,
|
"xfloat": None,
|
||||||
"xbool": None,
|
"xbool": None,
|
||||||
|
@ -107,12 +94,12 @@ class KustoClientTests(unittest.TestCase):
|
||||||
"xuint32": None,
|
"xuint32": None,
|
||||||
"xuint64": None,
|
"xuint64": None,
|
||||||
"xdate": None,
|
"xdate": None,
|
||||||
"xsmalltext": text_type(""),
|
"xsmalltext": str(""),
|
||||||
"xtext": text_type(""),
|
"xtext": str(""),
|
||||||
"xnumberAsText": text_type(""),
|
"xnumberAsText": str(""),
|
||||||
"xtime": None,
|
"xtime": None,
|
||||||
"xtextWithNulls": text_type(""),
|
"xtextWithNulls": str(""),
|
||||||
"xdynamicWithNulls": text_type(""),
|
"xdynamicWithNulls": str(""),
|
||||||
}
|
}
|
||||||
|
|
||||||
for row in response.primary_results[0]:
|
for row in response.primary_results[0]:
|
||||||
|
@ -157,7 +144,7 @@ class KustoClientTests(unittest.TestCase):
|
||||||
self.assertEqual(type(row["xdynamicWithNulls"]), type(expected["xdynamicWithNulls"]))
|
self.assertEqual(type(row["xdynamicWithNulls"]), type(expected["xdynamicWithNulls"]))
|
||||||
|
|
||||||
expected["rownumber"] = 0 if expected["rownumber"] is None else expected["rownumber"] + 1
|
expected["rownumber"] = 0 if expected["rownumber"] is None else expected["rownumber"] + 1
|
||||||
expected["rowguid"] = text_type("0000000{0}-0000-0000-0001-020304050607".format(expected["rownumber"]))
|
expected["rowguid"] = str("0000000{0}-0000-0000-0001-020304050607".format(expected["rownumber"]))
|
||||||
expected["xdouble"] = round(float(0) if expected["xdouble"] is None else expected["xdouble"] + 1.0001, 4)
|
expected["xdouble"] = round(float(0) if expected["xdouble"] is None else expected["xdouble"] + 1.0001, 4)
|
||||||
expected["xfloat"] = round(float(0) if expected["xfloat"] is None else expected["xfloat"] + 1.01, 2)
|
expected["xfloat"] = round(float(0) if expected["xfloat"] is None else expected["xfloat"] + 1.01, 2)
|
||||||
expected["xbool"] = False if expected["xbool"] is None else not expected["xbool"]
|
expected["xbool"] = False if expected["xbool"] is None else not expected["xbool"]
|
||||||
|
@ -172,7 +159,7 @@ class KustoClientTests(unittest.TestCase):
|
||||||
expected["xdate"] = expected["xdate"].replace(year=expected["xdate"].year + 1)
|
expected["xdate"] = expected["xdate"].replace(year=expected["xdate"].year + 1)
|
||||||
expected["xsmalltext"] = DIGIT_WORDS[int(expected["xint16"])]
|
expected["xsmalltext"] = DIGIT_WORDS[int(expected["xint16"])]
|
||||||
expected["xtext"] = DIGIT_WORDS[int(expected["xint16"])]
|
expected["xtext"] = DIGIT_WORDS[int(expected["xint16"])]
|
||||||
expected["xnumberAsText"] = text_type(expected["xint16"])
|
expected["xnumberAsText"] = str(expected["xint16"])
|
||||||
|
|
||||||
next_time = (
|
next_time = (
|
||||||
timedelta()
|
timedelta()
|
||||||
|
@ -281,8 +268,8 @@ class KustoClientTests(unittest.TestCase):
|
||||||
"xtextWithNulls": Series(["", "", "", "", "", "", "", "", "", "", ""], dtype=object),
|
"xtextWithNulls": Series(["", "", "", "", "", "", "", "", "", "", ""], dtype=object),
|
||||||
"xdynamicWithNulls": Series(
|
"xdynamicWithNulls": Series(
|
||||||
[
|
[
|
||||||
text_type(""),
|
str(""),
|
||||||
text_type(""),
|
str(""),
|
||||||
{"rowId": 1, "arr": [0, 1]},
|
{"rowId": 1, "arr": [0, 1]},
|
||||||
{"rowId": 2, "arr": [0, 2]},
|
{"rowId": 2, "arr": [0, 2]},
|
||||||
{"rowId": 3, "arr": [0, 3]},
|
{"rowId": 3, "arr": [0, 3]},
|
||||||
|
@ -363,15 +350,15 @@ range x from 1 to 10 step 1"""
|
||||||
self.assertIsInstance(row[0], int)
|
self.assertIsInstance(row[0], int)
|
||||||
self.assertEqual(row[0], 123)
|
self.assertEqual(row[0], 123)
|
||||||
|
|
||||||
self.assertIsInstance(row[1], text_type)
|
self.assertIsInstance(row[1], str)
|
||||||
self.assertEqual(row[1], "123")
|
self.assertEqual(row[1], "123")
|
||||||
|
|
||||||
self.assertIsInstance(row[2], text_type)
|
self.assertIsInstance(row[2], str)
|
||||||
self.assertEqual(row[2], "test bad json")
|
self.assertEqual(row[2], "test bad json")
|
||||||
|
|
||||||
self.assertEqual(row[3], None)
|
self.assertEqual(row[3], None)
|
||||||
|
|
||||||
self.assertIsInstance(row[4], text_type)
|
self.assertIsInstance(row[4], str)
|
||||||
self.assertEqual(row[4], '{"rowId":2,"arr":[0,2]}')
|
self.assertEqual(row[4], '{"rowId":2,"arr":[0,2]}')
|
||||||
|
|
||||||
self.assertIsInstance(row[5], dict)
|
self.assertIsInstance(row[5], dict)
|
||||||
|
|
|
@ -21,15 +21,15 @@ class KustoConnectionStringBuilderTests(unittest.TestCase):
|
||||||
]
|
]
|
||||||
|
|
||||||
for kcsb in kcsbs:
|
for kcsb in kcsbs:
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertFalse(kcsb.aad_federated_security)
|
assert not kcsb.aad_federated_security
|
||||||
self.assertIsNone(kcsb.aad_user_id)
|
assert kcsb.aad_user_id is None
|
||||||
self.assertIsNone(kcsb.password)
|
assert kcsb.password is None
|
||||||
self.assertIsNone(kcsb.application_client_id)
|
assert kcsb.application_client_id is None
|
||||||
self.assertIsNone(kcsb.application_key)
|
assert kcsb.application_key is None
|
||||||
self.assertEqual(kcsb.authority_id, "common")
|
assert kcsb.authority_id == "common"
|
||||||
self.assertEqual(repr(kcsb), "Data Source=localhost;Authority Id=common")
|
assert repr(kcsb) == "Data Source=localhost;Authority Id=common"
|
||||||
self.assertEqual(str(kcsb), "Data Source=localhost;Authority Id=common")
|
assert str(kcsb) == "Data Source=localhost;Authority Id=common"
|
||||||
|
|
||||||
def test_aad_app(self):
|
def test_aad_app(self):
|
||||||
"""Checks kcsb that is created with AAD application credentials."""
|
"""Checks kcsb that is created with AAD application credentials."""
|
||||||
|
@ -78,24 +78,19 @@ class KustoConnectionStringBuilderTests(unittest.TestCase):
|
||||||
kcsbs.append(kcsb2)
|
kcsbs.append(kcsb2)
|
||||||
|
|
||||||
for kcsb in kcsbs:
|
for kcsb in kcsbs:
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertTrue(kcsb.aad_federated_security)
|
assert kcsb.aad_federated_security
|
||||||
self.assertIsNone(kcsb.aad_user_id)
|
assert kcsb.aad_user_id is None
|
||||||
self.assertIsNone(kcsb.password)
|
assert kcsb.password is None
|
||||||
self.assertEqual(kcsb.application_client_id, uuid)
|
assert kcsb.application_client_id == uuid
|
||||||
self.assertEqual(kcsb.application_key, key)
|
assert kcsb.application_key == key
|
||||||
self.assertEqual(kcsb.authority_id, "microsoft.com")
|
assert kcsb.authority_id == "microsoft.com"
|
||||||
self.assertEqual(
|
assert repr(kcsb) == "Data Source=localhost;AAD Federated Security=True;Application Client Id={0};Application Key={1};Authority Id={2}".format(
|
||||||
repr(kcsb),
|
uuid, key, "microsoft.com"
|
||||||
"Data Source=localhost;AAD Federated Security=True;Application Client Id={0};Application Key={1};Authority Id={2}".format(
|
|
||||||
uuid, key, "microsoft.com"
|
|
||||||
),
|
|
||||||
)
|
)
|
||||||
self.assertEqual(
|
|
||||||
str(kcsb),
|
assert str(kcsb) == "Data Source=localhost;AAD Federated Security=True;Application Client Id={0};Application Key={1};Authority Id={2}".format(
|
||||||
"Data Source=localhost;AAD Federated Security=True;Application Client Id={0};Application Key={1};Authority Id={2}".format(
|
uuid, self.PASSWORDS_REPLACEMENT, "microsoft.com"
|
||||||
uuid, self.PASSWORDS_REPLACEMENT, "microsoft.com"
|
|
||||||
),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_aad_user(self):
|
def test_aad_user(self):
|
||||||
|
@ -123,19 +118,16 @@ class KustoConnectionStringBuilderTests(unittest.TestCase):
|
||||||
kcsbs.append(kcsb2)
|
kcsbs.append(kcsb2)
|
||||||
|
|
||||||
for kcsb in kcsbs:
|
for kcsb in kcsbs:
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertTrue(kcsb.aad_federated_security)
|
assert kcsb.aad_federated_security
|
||||||
self.assertEqual(kcsb.aad_user_id, user)
|
assert kcsb.aad_user_id == user
|
||||||
self.assertEqual(kcsb.password, password)
|
assert kcsb.password == password
|
||||||
self.assertIsNone(kcsb.application_client_id)
|
assert kcsb.application_client_id is None
|
||||||
self.assertIsNone(kcsb.application_key)
|
assert kcsb.application_key is None
|
||||||
self.assertEqual(kcsb.authority_id, "common")
|
assert kcsb.authority_id == "common"
|
||||||
self.assertEqual(
|
assert repr(kcsb) == "Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=common".format(user, password)
|
||||||
repr(kcsb), "Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=common".format(user, password),
|
assert str(kcsb) == "Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=common".format(
|
||||||
)
|
user, self.PASSWORDS_REPLACEMENT
|
||||||
self.assertEqual(
|
|
||||||
str(kcsb),
|
|
||||||
"Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=common".format(user, self.PASSWORDS_REPLACEMENT),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_aad_user_with_authority(self):
|
def test_aad_user_with_authority(self):
|
||||||
|
@ -146,71 +138,62 @@ class KustoConnectionStringBuilderTests(unittest.TestCase):
|
||||||
|
|
||||||
kcsb = KustoConnectionStringBuilder.with_aad_user_password_authentication("localhost", user, password, authority_id)
|
kcsb = KustoConnectionStringBuilder.with_aad_user_password_authentication("localhost", user, password, authority_id)
|
||||||
|
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertTrue(kcsb.aad_federated_security)
|
assert kcsb.aad_federated_security
|
||||||
self.assertEqual(kcsb.aad_user_id, user)
|
assert kcsb.aad_user_id == user
|
||||||
self.assertEqual(kcsb.password, password)
|
assert kcsb.password == password
|
||||||
self.assertIsNone(kcsb.application_client_id)
|
assert kcsb.application_client_id is None
|
||||||
self.assertIsNone(kcsb.application_key)
|
assert kcsb.application_key is None
|
||||||
self.assertEqual(kcsb.authority_id, authority_id)
|
assert kcsb.authority_id == authority_id
|
||||||
self.assertEqual(
|
assert repr(kcsb) == "Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=13456".format(user, password)
|
||||||
repr(kcsb), "Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=13456".format(user, password),
|
assert str(kcsb) == "Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=13456".format(
|
||||||
)
|
user, self.PASSWORDS_REPLACEMENT
|
||||||
self.assertEqual(
|
|
||||||
str(kcsb),
|
|
||||||
"Data Source=localhost;AAD Federated Security=True;AAD User ID={0};Password={1};Authority Id=13456".format(user, self.PASSWORDS_REPLACEMENT),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_aad_device_login(self):
|
def test_aad_device_login(self):
|
||||||
"""Checks kcsb that is created with AAD device login."""
|
"""Checks kcsb that is created with AAD device login."""
|
||||||
kcsb = KustoConnectionStringBuilder.with_aad_device_authentication("localhost")
|
kcsb = KustoConnectionStringBuilder.with_aad_device_authentication("localhost")
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertTrue(kcsb.aad_federated_security)
|
assert kcsb.aad_federated_security
|
||||||
self.assertIsNone(kcsb.aad_user_id)
|
assert kcsb.aad_user_id is None
|
||||||
self.assertIsNone(kcsb.password)
|
assert kcsb.password is None
|
||||||
self.assertIsNone(kcsb.application_client_id)
|
assert kcsb.application_client_id is None
|
||||||
self.assertIsNone(kcsb.application_key)
|
assert kcsb.application_key is None
|
||||||
self.assertEqual(kcsb.authority_id, "common")
|
assert kcsb.authority_id == "common"
|
||||||
self.assertEqual(repr(kcsb), "Data Source=localhost;AAD Federated Security=True;Authority Id=common")
|
assert repr(kcsb) == "Data Source=localhost;AAD Federated Security=True;Authority Id=common"
|
||||||
self.assertEqual(str(kcsb), "Data Source=localhost;AAD Federated Security=True;Authority Id=common")
|
assert str(kcsb) == "Data Source=localhost;AAD Federated Security=True;Authority Id=common"
|
||||||
|
|
||||||
def test_aad_app_token(self):
|
def test_aad_app_token(self):
|
||||||
"""Checks kcsb that is created with AAD user token."""
|
"""Checks kcsb that is created with AAD user token."""
|
||||||
token = "The app hardest token ever"
|
token = "The app hardest token ever"
|
||||||
kcsb = KustoConnectionStringBuilder.with_aad_application_token_authentication("localhost", application_token=token)
|
kcsb = KustoConnectionStringBuilder.with_aad_application_token_authentication("localhost", application_token=token)
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertEqual(kcsb.application_token, token)
|
assert kcsb.application_token == token
|
||||||
self.assertTrue(kcsb.aad_federated_security)
|
assert kcsb.aad_federated_security
|
||||||
self.assertIsNone(kcsb.aad_user_id)
|
assert kcsb.aad_user_id is None
|
||||||
self.assertIsNone(kcsb.password)
|
assert kcsb.password is None
|
||||||
self.assertIsNone(kcsb.application_client_id)
|
assert kcsb.application_client_id is None
|
||||||
self.assertIsNone(kcsb.application_key)
|
assert kcsb.application_key is None
|
||||||
self.assertIsNone(kcsb.user_token)
|
assert kcsb.user_token is None
|
||||||
self.assertEqual(kcsb.authority_id, "common")
|
assert kcsb.authority_id == "common"
|
||||||
self.assertEqual(
|
assert repr(kcsb) == "Data Source=localhost;AAD Federated Security=True;Authority Id=common;Application Token=%s" % token
|
||||||
repr(kcsb), "Data Source=localhost;AAD Federated Security=True;Authority Id=common;Application Token=%s" % token,
|
assert str(kcsb) == "Data Source=localhost;AAD Federated Security=True;Authority Id=common;Application Token=%s" % self.PASSWORDS_REPLACEMENT
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
str(kcsb), "Data Source=localhost;AAD Federated Security=True;Authority Id=common;Application Token=%s" % self.PASSWORDS_REPLACEMENT,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_aad_user_token(self):
|
def test_aad_user_token(self):
|
||||||
"""Checks kcsb that is created with AAD user token."""
|
"""Checks kcsb that is created with AAD user token."""
|
||||||
token = "The user hardest token ever"
|
token = "The user hardest token ever"
|
||||||
kcsb = KustoConnectionStringBuilder.with_aad_user_token_authentication("localhost", user_token=token)
|
kcsb = KustoConnectionStringBuilder.with_aad_user_token_authentication("localhost", user_token=token)
|
||||||
self.assertEqual(kcsb.data_source, "localhost")
|
assert kcsb.data_source == "localhost"
|
||||||
self.assertEqual(kcsb.user_token, token)
|
assert kcsb.user_token == token
|
||||||
self.assertTrue(kcsb.aad_federated_security)
|
assert kcsb.aad_federated_security
|
||||||
self.assertIsNone(kcsb.aad_user_id)
|
assert kcsb.aad_user_id is None
|
||||||
self.assertIsNone(kcsb.password)
|
assert kcsb.password is None
|
||||||
self.assertIsNone(kcsb.application_client_id)
|
assert kcsb.application_client_id is None
|
||||||
self.assertIsNone(kcsb.application_key)
|
assert kcsb.application_key is None
|
||||||
self.assertIsNone(kcsb.application_token)
|
assert kcsb.application_token is None
|
||||||
self.assertEqual(kcsb.authority_id, "common")
|
assert kcsb.authority_id == "common"
|
||||||
self.assertEqual(repr(kcsb), "Data Source=localhost;AAD Federated Security=True;Authority Id=common;User Token=%s" % token)
|
assert repr(kcsb) == "Data Source=localhost;AAD Federated Security=True;Authority Id=common;User Token=%s" % token
|
||||||
self.assertEqual(
|
assert str(kcsb) == "Data Source=localhost;AAD Federated Security=True;Authority Id=common;User Token=%s" % self.PASSWORDS_REPLACEMENT
|
||||||
str(kcsb), "Data Source=localhost;AAD Federated Security=True;Authority Id=common;User Token=%s" % self.PASSWORDS_REPLACEMENT,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_add_msi(self):
|
def test_add_msi(self):
|
||||||
client_guid = "kjhjk"
|
client_guid = "kjhjk"
|
||||||
|
|
|
@ -33,12 +33,7 @@ def test_msi_auth():
|
||||||
KustoConnectionStringBuilder.with_aad_managed_service_identity_authentication("localhost", msi_res_id=res_guid, timeout=1),
|
KustoConnectionStringBuilder.with_aad_managed_service_identity_authentication("localhost", msi_res_id=res_guid, timeout=1),
|
||||||
]
|
]
|
||||||
|
|
||||||
helpers = [
|
helpers = [_AadHelper(kcsb[0]), _AadHelper(kcsb[1]), _AadHelper(kcsb[2]), _AadHelper(kcsb[3])]
|
||||||
_AadHelper(kcsb[0]),
|
|
||||||
_AadHelper(kcsb[1]),
|
|
||||||
_AadHelper(kcsb[2]),
|
|
||||||
_AadHelper(kcsb[3]),
|
|
||||||
]
|
|
||||||
|
|
||||||
"""
|
"""
|
||||||
* * * Note * * *
|
* * * Note * * *
|
||||||
|
|
|
@ -82,7 +82,7 @@ class KustoIngestClient(object):
|
||||||
blob_service.create_blob_from_stream(container_name=container_details.object_name, blob_name=blob_name, stream=stream)
|
blob_service.create_blob_from_stream(container_name=container_details.object_name, blob_name=blob_name, stream=stream)
|
||||||
url = blob_service.make_blob_url(container_details.object_name, blob_name, sas_token=container_details.sas)
|
url = blob_service.make_blob_url(container_details.object_name, blob_name, sas_token=container_details.sas)
|
||||||
|
|
||||||
self.ingest_from_blob(BlobDescriptor(url, descriptor.size, descriptor.source_id,), ingestion_properties=ingestion_properties)
|
self.ingest_from_blob(BlobDescriptor(url, descriptor.size, descriptor.source_id), ingestion_properties=ingestion_properties)
|
||||||
|
|
||||||
def ingest_from_blob(self, blob_descriptor, ingestion_properties):
|
def ingest_from_blob(self, blob_descriptor, ingestion_properties):
|
||||||
"""Enqueuing an ingest command from azure blobs.
|
"""Enqueuing an ingest command from azure blobs.
|
||||||
|
|
|
@ -3,7 +3,6 @@
|
||||||
import json
|
import json
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from six import text_type
|
|
||||||
|
|
||||||
from ._descriptors import BlobDescriptor
|
from ._descriptors import BlobDescriptor
|
||||||
|
|
||||||
|
@ -22,7 +21,7 @@ class _IngestionBlobInfo:
|
||||||
self.properties["ReportMethod"] = ingestion_properties.report_method.value
|
self.properties["ReportMethod"] = ingestion_properties.report_method.value
|
||||||
self.properties["SourceMessageCreationTime"] = datetime.utcnow().isoformat()
|
self.properties["SourceMessageCreationTime"] = datetime.utcnow().isoformat()
|
||||||
self.properties["Id"] = (
|
self.properties["Id"] = (
|
||||||
text_type(blob_descriptor.source_id) if hasattr(blob_descriptor, "source_id") and blob_descriptor.source_id is not None else text_type(uuid.uuid4())
|
str(blob_descriptor.source_id) if hasattr(blob_descriptor, "source_id") and blob_descriptor.source_id is not None else str(uuid.uuid4())
|
||||||
)
|
)
|
||||||
|
|
||||||
additional_properties = ingestion_properties.additional_properties or {}
|
additional_properties = ingestion_properties.additional_properties or {}
|
||||||
|
@ -67,4 +66,4 @@ def _convert_list_to_json(array):
|
||||||
|
|
||||||
def _convert_dict_to_json(array):
|
def _convert_dict_to_json(array):
|
||||||
""" Converts array to a json string """
|
""" Converts array to a json string """
|
||||||
return json.dumps(array, skipkeys=False, allow_nan=False, indent=None, separators=(",", ":"), sort_keys=True, default=lambda o: o.__dict__,)
|
return json.dumps(array, skipkeys=False, allow_nan=False, indent=None, separators=(",", ":"), sort_keys=True, default=lambda o: o.__dict__)
|
||||||
|
|
|
@ -3,11 +3,7 @@
|
||||||
from enum import Enum, IntEnum
|
from enum import Enum, IntEnum
|
||||||
import warnings
|
import warnings
|
||||||
|
|
||||||
from .exceptions import (
|
from .exceptions import KustoDuplicateMappingError, KustoDuplicateMappingReferenceError, KustoMappingAndMappingReferenceError
|
||||||
KustoDuplicateMappingError,
|
|
||||||
KustoDuplicateMappingReferenceError,
|
|
||||||
KustoMappingAndMappingReferenceError,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class DataFormat(Enum):
|
class DataFormat(Enum):
|
||||||
|
@ -54,9 +50,7 @@ class ValidationImplications(IntEnum):
|
||||||
class ValidationPolicy(object):
|
class ValidationPolicy(object):
|
||||||
"""Validation policy to ingest command."""
|
"""Validation policy to ingest command."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(self, validationOptions=ValidationOptions.DoNotValidate, validationImplications=ValidationImplications.BestEffort):
|
||||||
self, validationOptions=ValidationOptions.DoNotValidate, validationImplications=ValidationImplications.BestEffort,
|
|
||||||
):
|
|
||||||
self.ValidationOptions = validationOptions
|
self.ValidationOptions = validationOptions
|
||||||
self.ValidationImplications = validationImplications
|
self.ValidationImplications = validationImplications
|
||||||
|
|
||||||
|
|
|
@ -26,7 +26,7 @@ class _ResourceUri:
|
||||||
|
|
||||||
class _IngestClientResources(object):
|
class _IngestClientResources(object):
|
||||||
def __init__(
|
def __init__(
|
||||||
self, secured_ready_for_aggregation_queues=None, failed_ingestions_queues=None, successful_ingestions_queues=None, containers=None, status_tables=None,
|
self, secured_ready_for_aggregation_queues=None, failed_ingestions_queues=None, successful_ingestions_queues=None, containers=None, status_tables=None
|
||||||
):
|
):
|
||||||
self.secured_ready_for_aggregation_queues = secured_ready_for_aggregation_queues
|
self.secured_ready_for_aggregation_queues = secured_ready_for_aggregation_queues
|
||||||
self.failed_ingestions_queues = failed_ingestions_queues
|
self.failed_ingestions_queues = failed_ingestions_queues
|
||||||
|
@ -77,7 +77,7 @@ class _ResourceManager(object):
|
||||||
containers = self._get_resource_by_name(table, "TempStorage")
|
containers = self._get_resource_by_name(table, "TempStorage")
|
||||||
status_tables = self._get_resource_by_name(table, "IngestionsStatusTable")
|
status_tables = self._get_resource_by_name(table, "IngestionsStatusTable")
|
||||||
|
|
||||||
return _IngestClientResources(secured_ready_for_aggregation_queues, failed_ingestions_queues, successful_ingestions_queues, containers, status_tables,)
|
return _IngestClientResources(secured_ready_for_aggregation_queues, failed_ingestions_queues, successful_ingestions_queues, containers, status_tables)
|
||||||
|
|
||||||
def _refresh_authorization_context(self):
|
def _refresh_authorization_context(self):
|
||||||
if (
|
if (
|
||||||
|
|
|
@ -10,7 +10,6 @@ from ._descriptors import FileDescriptor, StreamDescriptor
|
||||||
from .exceptions import KustoMissingMappingReferenceError
|
from .exceptions import KustoMissingMappingReferenceError
|
||||||
from ._ingestion_properties import DataFormat
|
from ._ingestion_properties import DataFormat
|
||||||
from io import TextIOWrapper, BytesIO
|
from io import TextIOWrapper, BytesIO
|
||||||
from six import string_types, PY2
|
|
||||||
|
|
||||||
|
|
||||||
class KustoStreamingIngestClient(object):
|
class KustoStreamingIngestClient(object):
|
||||||
|
@ -93,8 +92,8 @@ class KustoStreamingIngestClient(object):
|
||||||
zipped_stream = BytesIO()
|
zipped_stream = BytesIO()
|
||||||
buffer = stream.read()
|
buffer = stream.read()
|
||||||
with GzipFile(filename="data", fileobj=zipped_stream, mode="wb") as f_out:
|
with GzipFile(filename="data", fileobj=zipped_stream, mode="wb") as f_out:
|
||||||
if isinstance(buffer, string_types):
|
if isinstance(buffer, str):
|
||||||
data = bytes(buffer) if PY2 else bytes(buffer, "utf-8")
|
data = bytes(buffer, "utf-8")
|
||||||
f_out.write(data)
|
f_out.write(data)
|
||||||
else:
|
else:
|
||||||
f_out.write(buffer)
|
f_out.write(buffer)
|
||||||
|
|
|
@ -3,7 +3,6 @@
|
||||||
import json
|
import json
|
||||||
import base64
|
import base64
|
||||||
|
|
||||||
import six
|
|
||||||
from azure.storage.common import CloudStorageAccount
|
from azure.storage.common import CloudStorageAccount
|
||||||
from ._status_q import StatusQueue
|
from ._status_q import StatusQueue
|
||||||
|
|
||||||
|
@ -22,7 +21,7 @@ class StatusMessage(object):
|
||||||
self._raw = s
|
self._raw = s
|
||||||
|
|
||||||
o = json.loads(s)
|
o = json.loads(s)
|
||||||
for key, value in six.iteritems(o):
|
for key, value in o.items():
|
||||||
if hasattr(self, key):
|
if hasattr(self, key):
|
||||||
try:
|
try:
|
||||||
setattr(self, key, value)
|
setattr(self, key, value)
|
||||||
|
|
|
@ -37,12 +37,6 @@ setup(
|
||||||
"License :: OSI Approved :: MIT License",
|
"License :: OSI Approved :: MIT License",
|
||||||
],
|
],
|
||||||
packages=find_packages(exclude=["azure", "tests"]),
|
packages=find_packages(exclude=["azure", "tests"]),
|
||||||
install_requires=[
|
install_requires=["azure-kusto-data>={}".format(VERSION), "azure-storage-blob==2.1.0", "azure-storage-common==2.1.0", "azure-storage-queue==2.1.0"],
|
||||||
"azure-kusto-data>={}".format(VERSION),
|
|
||||||
"azure-storage-blob==2.1.0",
|
|
||||||
"azure-storage-common==2.1.0",
|
|
||||||
"azure-storage-queue==2.1.0",
|
|
||||||
"six>=1.10.0",
|
|
||||||
],
|
|
||||||
extras_require={"pandas": ["pandas==0.24.1"], ":python_version<'3.0'": ["azure-nspkg"]},
|
extras_require={"pandas": ["pandas==0.24.1"], ":python_version<'3.0'": ["azure-nspkg"]},
|
||||||
)
|
)
|
||||||
|
|
|
@ -4,7 +4,6 @@ import time
|
||||||
import os
|
import os
|
||||||
import uuid
|
import uuid
|
||||||
import io
|
import io
|
||||||
from six import text_type
|
|
||||||
|
|
||||||
from azure.kusto.data.request import KustoClient, KustoConnectionStringBuilder
|
from azure.kusto.data.request import KustoClient, KustoConnectionStringBuilder
|
||||||
from azure.kusto.ingest.status import KustoIngestStatusQueues
|
from azure.kusto.ingest.status import KustoIngestStatusQueues
|
||||||
|
@ -103,7 +102,7 @@ client.execute(db_name, ".drop table {} ifexists".format(table_name))
|
||||||
@pytest.mark.run(order=1)
|
@pytest.mark.run(order=1)
|
||||||
def test_csv_ingest_non_existing_table():
|
def test_csv_ingest_non_existing_table():
|
||||||
csv_ingest_props = IngestionProperties(
|
csv_ingest_props = IngestionProperties(
|
||||||
db_name, table_name, dataFormat=DataFormat.CSV, mapping=Helpers.create_deft_table_csv_mappings(), reportLevel=ReportLevel.FailuresAndSuccesses,
|
db_name, table_name, dataFormat=DataFormat.CSV, mapping=Helpers.create_deft_table_csv_mappings(), reportLevel=ReportLevel.FailuresAndSuccesses
|
||||||
)
|
)
|
||||||
csv_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.csv")
|
csv_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.csv")
|
||||||
zipped_csv_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.csv.gz")
|
zipped_csv_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.csv.gz")
|
||||||
|
@ -130,7 +129,7 @@ def test_csv_ingest_non_existing_table():
|
||||||
time.sleep(20)
|
time.sleep(20)
|
||||||
response = client.execute(db_name, "{} | count".format(table_name))
|
response = client.execute(db_name, "{} | count".format(table_name))
|
||||||
for row in response.primary_results[0]:
|
for row in response.primary_results[0]:
|
||||||
assert int(row["Count"]) == 20, "{0} | count = {1}".format(table_name, text_type(row["Count"]))
|
assert int(row["Count"]) == 20, "{0} | count = {1}".format(table_name, str(row["Count"]))
|
||||||
|
|
||||||
|
|
||||||
json_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.json")
|
json_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.json")
|
||||||
|
@ -140,7 +139,7 @@ zipped_json_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests",
|
||||||
@pytest.mark.run(order=2)
|
@pytest.mark.run(order=2)
|
||||||
def test_json_ingest_existing_table():
|
def test_json_ingest_existing_table():
|
||||||
json_ingestion_props = IngestionProperties(
|
json_ingestion_props = IngestionProperties(
|
||||||
db_name, table_name, dataFormat=DataFormat.JSON, mapping=Helpers.create_deft_table_json_mappings(), reportLevel=ReportLevel.FailuresAndSuccesses,
|
db_name, table_name, dataFormat=DataFormat.JSON, mapping=Helpers.create_deft_table_json_mappings(), reportLevel=ReportLevel.FailuresAndSuccesses
|
||||||
)
|
)
|
||||||
|
|
||||||
for f in [json_file_path, zipped_json_file_path]:
|
for f in [json_file_path, zipped_json_file_path]:
|
||||||
|
@ -166,14 +165,14 @@ def test_json_ingest_existing_table():
|
||||||
time.sleep(20)
|
time.sleep(20)
|
||||||
response = client.execute(db_name, "{} | count".format(table_name))
|
response = client.execute(db_name, "{} | count".format(table_name))
|
||||||
for row in response.primary_results[0]:
|
for row in response.primary_results[0]:
|
||||||
assert int(row["Count"]) == 24, "{0} | count = {1}".format(table_name, text_type(row["Count"]))
|
assert int(row["Count"]) == 24, "{0} | count = {1}".format(table_name, str(row["Count"]))
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.run(order=3)
|
@pytest.mark.run(order=3)
|
||||||
def test_ingest_complicated_props():
|
def test_ingest_complicated_props():
|
||||||
# Test ingest with complicated ingestion properties
|
# Test ingest with complicated ingestion properties
|
||||||
validation_policy = ValidationPolicy(
|
validation_policy = ValidationPolicy(
|
||||||
validationOptions=ValidationOptions.ValidateCsvInputConstantColumns, validationImplications=ValidationImplications.Fail,
|
validationOptions=ValidationOptions.ValidateCsvInputConstantColumns, validationImplications=ValidationImplications.Fail
|
||||||
)
|
)
|
||||||
json_ingestion_props = IngestionProperties(
|
json_ingestion_props = IngestionProperties(
|
||||||
db_name,
|
db_name,
|
||||||
|
@ -216,7 +215,7 @@ def test_ingest_complicated_props():
|
||||||
time.sleep(20)
|
time.sleep(20)
|
||||||
response = client.execute(db_name, "{} | count".format(table_name))
|
response = client.execute(db_name, "{} | count".format(table_name))
|
||||||
for row in response.primary_results[0]:
|
for row in response.primary_results[0]:
|
||||||
assert int(row["Count"]) == 28, "{0} | count = {1}".format(table_name, text_type(row["Count"]))
|
assert int(row["Count"]) == 28, "{0} | count = {1}".format(table_name, str(row["Count"]))
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.run(order=4)
|
@pytest.mark.run(order=4)
|
||||||
|
@ -253,13 +252,13 @@ def test_json_ingestion_ingest_by_tag():
|
||||||
time.sleep(20)
|
time.sleep(20)
|
||||||
response = client.execute(db_name, "{} | count".format(table_name))
|
response = client.execute(db_name, "{} | count".format(table_name))
|
||||||
for row in response.primary_results[0]:
|
for row in response.primary_results[0]:
|
||||||
assert int(row["Count"]) == 28, "{0} | count = {1}".format(table_name, text_type(row["Count"]))
|
assert int(row["Count"]) == 28, "{0} | count = {1}".format(table_name, str(row["Count"]))
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.run(order=5)
|
@pytest.mark.run(order=5)
|
||||||
def test_tsv_ingestion_csv_mapping():
|
def test_tsv_ingestion_csv_mapping():
|
||||||
tsv_ingestion_props = IngestionProperties(
|
tsv_ingestion_props = IngestionProperties(
|
||||||
db_name, table_name, dataFormat=DataFormat.TSV, mapping=Helpers.create_deft_table_csv_mappings(), reportLevel=ReportLevel.FailuresAndSuccesses,
|
db_name, table_name, dataFormat=DataFormat.TSV, mapping=Helpers.create_deft_table_csv_mappings(), reportLevel=ReportLevel.FailuresAndSuccesses
|
||||||
)
|
)
|
||||||
tsv_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.tsv")
|
tsv_file_path = os.path.join(os.getcwd(), "azure-kusto-ingest", "tests", "input", "dataset.tsv")
|
||||||
|
|
||||||
|
@ -284,7 +283,7 @@ def test_tsv_ingestion_csv_mapping():
|
||||||
time.sleep(20)
|
time.sleep(20)
|
||||||
response = client.execute(db_name, "{} | count".format(table_name))
|
response = client.execute(db_name, "{} | count".format(table_name))
|
||||||
for row in response.primary_results[0]:
|
for row in response.primary_results[0]:
|
||||||
assert int(row["Count"]) == 38, "{0} | count = {1}".format(table_name, text_type(row["Count"]))
|
assert int(row["Count"]) == 38, "{0} | count = {1}".format(table_name, str(row["Count"]))
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.run(order=6)
|
@pytest.mark.run(order=6)
|
||||||
|
@ -431,7 +430,7 @@ def test_streaming_ingest_from_dataframe():
|
||||||
"xtextWithNulls",
|
"xtextWithNulls",
|
||||||
"xdynamicWithNulls",
|
"xdynamicWithNulls",
|
||||||
]
|
]
|
||||||
rows = [[0, "00000000-0000-0000-0001-020304050607", 0.0, 0.0, 0, 0, 0, 0, 0, 0, 0, 0, "2014-01-01T01:01:01Z", "Zero", "Zero", "0", "00:00:00", None, "",]]
|
rows = [[0, "00000000-0000-0000-0001-020304050607", 0.0, 0.0, 0, 0, 0, 0, 0, 0, 0, 0, "2014-01-01T01:01:01Z", "Zero", "Zero", "0", "00:00:00", None, ""]]
|
||||||
df = DataFrame(data=rows, columns=fields)
|
df = DataFrame(data=rows, columns=fields)
|
||||||
ingestion_properties = IngestionProperties(database=db_name, table=table_name, dataFormat=DataFormat.CSV)
|
ingestion_properties = IngestionProperties(database=db_name, table=table_name, dataFormat=DataFormat.CSV)
|
||||||
ingest_client.ingest_from_dataframe(df, ingestion_properties)
|
ingest_client.ingest_from_dataframe(df, ingestion_properties)
|
||||||
|
|
|
@ -15,10 +15,10 @@ class ResourceUriTests(unittest.TestCase):
|
||||||
|
|
||||||
uri = "https://{}.blob.core.windows.net/{}?{}".format(storage_name, container_name, container_sas)
|
uri = "https://{}.blob.core.windows.net/{}?{}".format(storage_name, container_name, container_sas)
|
||||||
connection_string = _ResourceUri.parse(uri)
|
connection_string = _ResourceUri.parse(uri)
|
||||||
self.assertEqual(connection_string.storage_account_name, storage_name)
|
assert connection_string.storage_account_name == storage_name
|
||||||
self.assertEqual(connection_string.object_type, "blob")
|
assert connection_string.object_type == "blob"
|
||||||
self.assertEqual(connection_string.sas, container_sas)
|
assert connection_string.sas == container_sas
|
||||||
self.assertEqual(connection_string.object_name, container_name)
|
assert connection_string.object_name == container_name
|
||||||
|
|
||||||
def test_queue_uri(self):
|
def test_queue_uri(self):
|
||||||
"""Tests parsing queues uris."""
|
"""Tests parsing queues uris."""
|
||||||
|
@ -28,7 +28,7 @@ class ResourceUriTests(unittest.TestCase):
|
||||||
|
|
||||||
uri = "https://{}.queue.core.windows.net/{}?{}".format(storage_name, queue_name, queue_sas)
|
uri = "https://{}.queue.core.windows.net/{}?{}".format(storage_name, queue_name, queue_sas)
|
||||||
connection_string = _ResourceUri.parse(uri)
|
connection_string = _ResourceUri.parse(uri)
|
||||||
self.assertEqual(connection_string.storage_account_name, storage_name)
|
assert connection_string.storage_account_name == storage_name
|
||||||
self.assertEqual(connection_string.object_type, "queue")
|
assert connection_string.object_type == "queue"
|
||||||
self.assertEqual(connection_string.sas, queue_sas)
|
assert connection_string.sas == queue_sas
|
||||||
self.assertEqual(connection_string.object_name, queue_name)
|
assert connection_string.object_name == queue_name
|
||||||
|
|
|
@ -12,62 +12,62 @@ class DescriptorsTest(unittest.TestCase):
|
||||||
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv")
|
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv")
|
||||||
descriptor = FileDescriptor(filePath, 10)
|
descriptor = FileDescriptor(filePath, 10)
|
||||||
with descriptor.open(True) as stream:
|
with descriptor.open(True) as stream:
|
||||||
self.assertEquals(descriptor.size, 10)
|
assert descriptor.size == 10
|
||||||
self.assertTrue(descriptor.stream_name.endswith(".csv.gz"))
|
assert descriptor.stream_name.endswith(".csv.gz")
|
||||||
if sys.version_info[0] >= 3:
|
if sys.version_info[0] >= 3:
|
||||||
self.assertTrue(stream.readable(), True)
|
assert stream.readable()
|
||||||
self.assertEqual(stream.tell(), 0)
|
assert stream.tell() == 0
|
||||||
|
|
||||||
self.assertEqual(stream.closed, True)
|
assert stream.closed == True
|
||||||
|
|
||||||
def test_unzipped_file_without_size(self):
|
def test_unzipped_file_without_size(self):
|
||||||
"""Tests FileDescriptor without size and unzipped file."""
|
"""Tests FileDescriptor without size and unzipped file."""
|
||||||
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv")
|
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv")
|
||||||
descriptor = FileDescriptor(filePath, 0)
|
descriptor = FileDescriptor(filePath, 0)
|
||||||
with descriptor.open(True) as stream:
|
with descriptor.open(True) as stream:
|
||||||
self.assertGreater(descriptor.size, 0)
|
assert descriptor.size > 0
|
||||||
self.assertTrue(descriptor.stream_name.endswith(".csv.gz"))
|
assert descriptor.stream_name.endswith(".csv.gz")
|
||||||
if sys.version_info[0] >= 3:
|
if sys.version_info[0] >= 3:
|
||||||
self.assertTrue(stream.readable(), True)
|
assert stream.readable()
|
||||||
self.assertEqual(stream.tell(), 0)
|
assert stream.tell() == 0
|
||||||
|
|
||||||
self.assertEqual(stream.closed, True)
|
assert stream.closed == True
|
||||||
|
|
||||||
def test_zipped_file_with_size(self):
|
def test_zipped_file_with_size(self):
|
||||||
"""Tests FileDescriptor with size and zipped file."""
|
"""Tests FileDescriptor with size and zipped file."""
|
||||||
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv.gz")
|
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv.gz")
|
||||||
descriptor = FileDescriptor(filePath, 10)
|
descriptor = FileDescriptor(filePath, 10)
|
||||||
with descriptor.open(False) as stream:
|
with descriptor.open(False) as stream:
|
||||||
self.assertGreater(descriptor.size, 10)
|
assert descriptor.size > 10
|
||||||
self.assertTrue(descriptor.stream_name.endswith(".csv.gz"))
|
assert descriptor.stream_name.endswith(".csv.gz")
|
||||||
if sys.version_info[0] >= 3:
|
if sys.version_info[0] >= 3:
|
||||||
self.assertTrue(stream.readable(), True)
|
assert stream.readable()
|
||||||
self.assertEqual(stream.tell(), 0)
|
assert stream.tell() == 0
|
||||||
|
|
||||||
self.assertEqual(stream.closed, True)
|
assert stream.closed == True
|
||||||
|
|
||||||
def test_zipped_file_without_size(self):
|
def test_zipped_file_without_size(self):
|
||||||
"""Tests FileDescriptor without size and zipped file."""
|
"""Tests FileDescriptor without size and zipped file."""
|
||||||
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv.gz")
|
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv.gz")
|
||||||
descriptor = FileDescriptor(filePath, 0)
|
descriptor = FileDescriptor(filePath, 0)
|
||||||
with descriptor.open(False) as stream:
|
with descriptor.open(False) as stream:
|
||||||
self.assertEqual(descriptor.size, 5071)
|
assert descriptor.size == 5071
|
||||||
self.assertTrue(descriptor.stream_name.endswith(".csv.gz"))
|
assert descriptor.stream_name.endswith(".csv.gz")
|
||||||
if sys.version_info[0] >= 3:
|
if sys.version_info[0] >= 3:
|
||||||
self.assertTrue(stream.readable(), True)
|
assert stream.readable()
|
||||||
self.assertEqual(stream.tell(), 0)
|
assert stream.tell() == 0
|
||||||
|
|
||||||
self.assertEqual(stream.closed, True)
|
assert stream.closed == True
|
||||||
|
|
||||||
def test_unzipped_file_dont_compress(self):
|
def test_unzipped_file_dont_compress(self):
|
||||||
"""Tests FileDescriptor with size and unzipped file."""
|
"""Tests FileDescriptor with size and unzipped file."""
|
||||||
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv")
|
filePath = path.join(path.dirname(path.abspath(__file__)), "input", "dataset.csv")
|
||||||
descriptor = FileDescriptor(filePath, 10)
|
descriptor = FileDescriptor(filePath, 10)
|
||||||
with descriptor.open(False) as stream:
|
with descriptor.open(False) as stream:
|
||||||
self.assertEqual(descriptor.size, 10)
|
assert descriptor.size == 10
|
||||||
self.assertTrue(descriptor.stream_name.endswith(".csv"))
|
assert descriptor.stream_name.endswith(".csv")
|
||||||
if sys.version_info[0] >= 3:
|
if sys.version_info[0] >= 3:
|
||||||
self.assertTrue(stream.readable(), True)
|
assert stream.readable()
|
||||||
self.assertEqual(stream.tell(), 0)
|
assert stream.tell() == 0
|
||||||
|
|
||||||
self.assertEqual(stream.closed, True)
|
assert stream.closed == True
|
||||||
|
|
|
@ -2,13 +2,8 @@ import unittest
|
||||||
import re
|
import re
|
||||||
import json
|
import json
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
from six import assertRegex
|
|
||||||
from azure.kusto.ingest._ingestion_blob_info import _IngestionBlobInfo
|
from azure.kusto.ingest._ingestion_blob_info import _IngestionBlobInfo
|
||||||
from azure.kusto.ingest.exceptions import (
|
from azure.kusto.ingest.exceptions import KustoDuplicateMappingError, KustoDuplicateMappingReferenceError, KustoMappingAndMappingReferenceError
|
||||||
KustoDuplicateMappingError,
|
|
||||||
KustoDuplicateMappingReferenceError,
|
|
||||||
KustoMappingAndMappingReferenceError,
|
|
||||||
)
|
|
||||||
from azure.kusto.ingest import (
|
from azure.kusto.ingest import (
|
||||||
BlobDescriptor,
|
BlobDescriptor,
|
||||||
IngestionProperties,
|
IngestionProperties,
|
||||||
|
@ -118,47 +113,37 @@ class IngestionBlobInfoTest(unittest.TestCase):
|
||||||
IngestionProperties(database="database", table="table", mapping="mapping", ingestionMapping="ingestionMapping")
|
IngestionProperties(database="database", table="table", mapping="mapping", ingestionMapping="ingestionMapping")
|
||||||
|
|
||||||
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
||||||
IngestionProperties(
|
IngestionProperties(database="database", table="table", mapping="mapping", ingestionMappingReference="ingestionMappingReference")
|
||||||
database="database", table="table", mapping="mapping", ingestionMappingReference="ingestionMappingReference",
|
|
||||||
)
|
|
||||||
|
|
||||||
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
||||||
IngestionProperties(
|
IngestionProperties(database="database", table="table", ingestionMapping="ingestionMapping", ingestionMappingReference="ingestionMappingReference")
|
||||||
database="database", table="table", ingestionMapping="ingestionMapping", ingestionMappingReference="ingestionMappingReference",
|
|
||||||
)
|
|
||||||
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
||||||
IngestionProperties(database="database", table="table", mapping="mapping", mappingReference="mappingReference")
|
IngestionProperties(database="database", table="table", mapping="mapping", mappingReference="mappingReference")
|
||||||
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
with self.assertRaises(KustoMappingAndMappingReferenceError):
|
||||||
IngestionProperties(
|
IngestionProperties(database="database", table="table", ingestionMapping="ingestionMapping", mappingReference="mappingReference")
|
||||||
database="database", table="table", ingestionMapping="ingestionMapping", mappingReference="mappingReference",
|
|
||||||
)
|
|
||||||
with self.assertRaises(KustoDuplicateMappingReferenceError):
|
with self.assertRaises(KustoDuplicateMappingReferenceError):
|
||||||
IngestionProperties(
|
IngestionProperties(database="database", table="table", mappingReference="mappingReference", ingestionMappingReference="ingestionMappingReference")
|
||||||
database="database", table="table", mappingReference="mappingReference", ingestionMappingReference="ingestionMappingReference",
|
|
||||||
)
|
|
||||||
|
|
||||||
def _verify_ingestion_blob_info_result(self, ingestion_blob_info):
|
def _verify_ingestion_blob_info_result(self, ingestion_blob_info):
|
||||||
result = json.loads(ingestion_blob_info)
|
result = json.loads(ingestion_blob_info)
|
||||||
self.assertIsNotNone(result)
|
assert result is not None
|
||||||
self.assertIsInstance(result, dict)
|
assert isinstance(result, dict)
|
||||||
self.assertEqual(result["BlobPath"], "somepath")
|
assert result["BlobPath"] == "somepath"
|
||||||
self.assertEqual(result["DatabaseName"], "database")
|
assert result["DatabaseName"] == "database"
|
||||||
self.assertEqual(result["TableName"], "table")
|
assert result["TableName"] == "table"
|
||||||
self.assertIsInstance(result["RawDataSize"], int)
|
assert isinstance(result["RawDataSize"], int)
|
||||||
self.assertIsInstance(result["IgnoreSizeLimit"], bool)
|
assert isinstance(result["IgnoreSizeLimit"], bool)
|
||||||
self.assertIsInstance(result["FlushImmediately"], bool)
|
assert isinstance(result["FlushImmediately"], bool)
|
||||||
self.assertIsInstance(result["RetainBlobOnSuccess"], bool)
|
assert isinstance(result["RetainBlobOnSuccess"], bool)
|
||||||
self.assertIsInstance(result["ReportMethod"], int)
|
assert isinstance(result["ReportMethod"], int)
|
||||||
self.assertIsInstance(result["ReportLevel"], int)
|
assert isinstance(result["ReportLevel"], int)
|
||||||
self.assertIsInstance(UUID(result["Id"]), UUID)
|
assert isinstance(UUID(result["Id"]), UUID)
|
||||||
assertRegex(self, result["SourceMessageCreationTime"], TIMESTAMP_REGEX)
|
assert re.match(TIMESTAMP_REGEX, result["SourceMessageCreationTime"])
|
||||||
self.assertEqual(result["AdditionalProperties"]["authorizationContext"], "authorizationContextText")
|
assert result["AdditionalProperties"]["authorizationContext"] == "authorizationContextText"
|
||||||
self.assertEqual(result["AdditionalProperties"]["ingestIfNotExists"], '["ingestIfNotExistTags"]')
|
assert result["AdditionalProperties"]["ingestIfNotExists"] == '["ingestIfNotExistTags"]'
|
||||||
self.assertIn(
|
assert result["AdditionalProperties"]["ValidationPolicy"] in (
|
||||||
result["AdditionalProperties"]["ValidationPolicy"],
|
'{"ValidationImplications":1,"ValidationOptions":1}',
|
||||||
(
|
'{"ValidationImplications":ValidationImplications.BestEffort,"ValidationOptions":ValidationOptions.ValidateCsvInputConstantColumns}',
|
||||||
'{"ValidationImplications":1,"ValidationOptions":1}',
|
|
||||||
'{"ValidationImplications":ValidationImplications.BestEffort,"ValidationOptions":ValidationOptions.ValidateCsvInputConstantColumns}',
|
|
||||||
),
|
|
||||||
)
|
)
|
||||||
self.assertEqual(result["AdditionalProperties"]["tags"], '["tag","drop-by:dropByTags","ingest-by:ingestByTags"]')
|
|
||||||
|
assert result["AdditionalProperties"]["tags"] == '["tag","drop-by:dropByTags","ingest-by:ingestByTags"]'
|
||||||
|
|
|
@ -35,15 +35,15 @@ def request_callback(request):
|
||||||
"Tables": [
|
"Tables": [
|
||||||
{
|
{
|
||||||
"TableName": "Table_0",
|
"TableName": "Table_0",
|
||||||
"Columns": [{"ColumnName": "ResourceTypeName", "DataType": "String"}, {"ColumnName": "StorageRoot", "DataType": "String"},],
|
"Columns": [{"ColumnName": "ResourceTypeName", "DataType": "String"}, {"ColumnName": "StorageRoot", "DataType": "String"}],
|
||||||
"Rows": [
|
"Rows": [
|
||||||
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas",],
|
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas"],
|
||||||
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas",],
|
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas"],
|
||||||
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas",],
|
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas"],
|
||||||
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas",],
|
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas"],
|
||||||
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas",],
|
["SecuredReadyForAggregationQueue", "https://storageaccount.queue.core.windows.net/readyforaggregation-secured?sas"],
|
||||||
["FailedIngestionsQueue", "https://storageaccount.queue.core.windows.net/failedingestions?sas"],
|
["FailedIngestionsQueue", "https://storageaccount.queue.core.windows.net/failedingestions?sas"],
|
||||||
["SuccessfulIngestionsQueue", "https://storageaccount.queue.core.windows.net/successfulingestions?sas",],
|
["SuccessfulIngestionsQueue", "https://storageaccount.queue.core.windows.net/successfulingestions?sas"],
|
||||||
["TempStorage", "https://storageaccount.blob.core.windows.net/tempstorage?sas"],
|
["TempStorage", "https://storageaccount.blob.core.windows.net/tempstorage?sas"],
|
||||||
["TempStorage", "https://storageaccount.blob.core.windows.net/tempstorage?sas"],
|
["TempStorage", "https://storageaccount.blob.core.windows.net/tempstorage?sas"],
|
||||||
["TempStorage", "https://storageaccount.blob.core.windows.net/tempstorage?sas"],
|
["TempStorage", "https://storageaccount.blob.core.windows.net/tempstorage?sas"],
|
||||||
|
@ -58,9 +58,7 @@ def request_callback(request):
|
||||||
if ".get kusto identity token" in body["csl"]:
|
if ".get kusto identity token" in body["csl"]:
|
||||||
response_status = 200
|
response_status = 200
|
||||||
response_body = {
|
response_body = {
|
||||||
"Tables": [
|
"Tables": [{"TableName": "Table_0", "Columns": [{"ColumnName": "AuthorizationContext", "DataType": "String"}], "Rows": [["authorization_context"]]}]
|
||||||
{"TableName": "Table_0", "Columns": [{"ColumnName": "AuthorizationContext", "DataType": "String"}], "Rows": [["authorization_context"]],}
|
|
||||||
]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return (response_status, response_headers, json.dumps(response_body))
|
return (response_status, response_headers, json.dumps(response_body))
|
||||||
|
@ -78,7 +76,7 @@ class KustoIngestClientTests(unittest.TestCase):
|
||||||
@patch("uuid.uuid4", return_value=MOCKED_UUID_4)
|
@patch("uuid.uuid4", return_value=MOCKED_UUID_4)
|
||||||
def test_sanity_ingest_from_file(self, mock_uuid, mock_put_message_in_queue, mock_create_blob_from_stream, mock_aad):
|
def test_sanity_ingest_from_file(self, mock_uuid, mock_put_message_in_queue, mock_create_blob_from_stream, mock_aad):
|
||||||
responses.add_callback(
|
responses.add_callback(
|
||||||
responses.POST, "https://ingest-somecluster.kusto.windows.net/v1/rest/mgmt", callback=request_callback, content_type="application/json",
|
responses.POST, "https://ingest-somecluster.kusto.windows.net/v1/rest/mgmt", callback=request_callback, content_type="application/json"
|
||||||
)
|
)
|
||||||
|
|
||||||
ingest_client = KustoIngestClient("https://ingest-somecluster.kusto.windows.net")
|
ingest_client = KustoIngestClient("https://ingest-somecluster.kusto.windows.net")
|
||||||
|
@ -130,7 +128,7 @@ class KustoIngestClientTests(unittest.TestCase):
|
||||||
@patch("os.getpid", return_value=MOCKED_PID)
|
@patch("os.getpid", return_value=MOCKED_PID)
|
||||||
def test_simple_ingest_from_dataframe(self, mock_pid, mock_time, mock_uuid, mock_put_message_in_queue, mock_create_blob_from_stream):
|
def test_simple_ingest_from_dataframe(self, mock_pid, mock_time, mock_uuid, mock_put_message_in_queue, mock_create_blob_from_stream):
|
||||||
responses.add_callback(
|
responses.add_callback(
|
||||||
responses.POST, "https://ingest-somecluster.kusto.windows.net/v1/rest/mgmt", callback=request_callback, content_type="application/json",
|
responses.POST, "https://ingest-somecluster.kusto.windows.net/v1/rest/mgmt", callback=request_callback, content_type="application/json"
|
||||||
)
|
)
|
||||||
|
|
||||||
ingest_client = KustoIngestClient("https://ingest-somecluster.kusto.windows.net")
|
ingest_client = KustoIngestClient("https://ingest-somecluster.kusto.windows.net")
|
||||||
|
|
|
@ -46,9 +46,7 @@ def request_callback(request):
|
||||||
class KustoStreamingIngestClientTests(unittest.TestCase):
|
class KustoStreamingIngestClientTests(unittest.TestCase):
|
||||||
@responses.activate
|
@responses.activate
|
||||||
def test_streaming_ingest_from_file(self):
|
def test_streaming_ingest_from_file(self):
|
||||||
responses.add_callback(
|
responses.add_callback(responses.POST, "https://somecluster.kusto.windows.net/v1/rest/ingest/database/table", callback=request_callback)
|
||||||
responses.POST, "https://somecluster.kusto.windows.net/v1/rest/ingest/database/table", callback=request_callback,
|
|
||||||
)
|
|
||||||
|
|
||||||
ingest_client = KustoStreamingIngestClient("https://somecluster.kusto.windows.net")
|
ingest_client = KustoStreamingIngestClient("https://somecluster.kusto.windows.net")
|
||||||
ingestion_properties = IngestionProperties(database="database", table="table", dataFormat=DataFormat.CSV)
|
ingestion_properties = IngestionProperties(database="database", table="table", dataFormat=DataFormat.CSV)
|
||||||
|
@ -112,9 +110,7 @@ class KustoStreamingIngestClientTests(unittest.TestCase):
|
||||||
@pytest.mark.skipif(not pandas_installed, reason="requires pandas")
|
@pytest.mark.skipif(not pandas_installed, reason="requires pandas")
|
||||||
@responses.activate
|
@responses.activate
|
||||||
def test_streaming_ingest_from_dataframe(self):
|
def test_streaming_ingest_from_dataframe(self):
|
||||||
responses.add_callback(
|
responses.add_callback(responses.POST, "https://somecluster.kusto.windows.net/v1/rest/ingest/database/table", callback=request_callback)
|
||||||
responses.POST, "https://somecluster.kusto.windows.net/v1/rest/ingest/database/table", callback=request_callback,
|
|
||||||
)
|
|
||||||
|
|
||||||
ingest_client = KustoStreamingIngestClient("https://somecluster.kusto.windows.net")
|
ingest_client = KustoStreamingIngestClient("https://somecluster.kusto.windows.net")
|
||||||
ingestion_properties = IngestionProperties(database="database", table="table", dataFormat=DataFormat.CSV)
|
ingestion_properties = IngestionProperties(database="database", table="table", dataFormat=DataFormat.CSV)
|
||||||
|
@ -129,9 +125,7 @@ class KustoStreamingIngestClientTests(unittest.TestCase):
|
||||||
|
|
||||||
@responses.activate
|
@responses.activate
|
||||||
def test_streaming_ingest_from_stream(self):
|
def test_streaming_ingest_from_stream(self):
|
||||||
responses.add_callback(
|
responses.add_callback(responses.POST, "https://somecluster.kusto.windows.net/v1/rest/ingest/database/table", callback=request_callback)
|
||||||
responses.POST, "https://somecluster.kusto.windows.net/v1/rest/ingest/database/table", callback=request_callback,
|
|
||||||
)
|
|
||||||
|
|
||||||
ingest_client = KustoStreamingIngestClient("https://somecluster.kusto.windows.net")
|
ingest_client = KustoStreamingIngestClient("https://somecluster.kusto.windows.net")
|
||||||
ingestion_properties = IngestionProperties(database="database", table="table", dataFormat=DataFormat.CSV)
|
ingestion_properties = IngestionProperties(database="database", table="table", dataFormat=DataFormat.CSV)
|
||||||
|
|
|
@ -199,10 +199,10 @@ class StatusQTests(unittest.TestCase):
|
||||||
assert len(get_failure_actual) == 6
|
assert len(get_failure_actual) == 6
|
||||||
|
|
||||||
for m in get_failure_actual:
|
for m in get_failure_actual:
|
||||||
assert isinstance(m, FailureMessage) == True
|
assert isinstance(m, FailureMessage)
|
||||||
|
|
||||||
for m in get_success_actual:
|
for m in get_success_actual:
|
||||||
assert isinstance(m, SuccessMessage) == True
|
assert isinstance(m, SuccessMessage)
|
||||||
|
|
||||||
assert mocked_q_get_messages.call_count == 3
|
assert mocked_q_get_messages.call_count == 3
|
||||||
assert mocked_q_del_messages.call_count == len(get_success_actual) + len(get_failure_actual)
|
assert mocked_q_del_messages.call_count == len(get_success_actual) + len(get_failure_actual)
|
||||||
|
@ -285,7 +285,7 @@ class StatusQTests(unittest.TestCase):
|
||||||
assert len(get_failure_actual) == 6
|
assert len(get_failure_actual) == 6
|
||||||
|
|
||||||
for m in get_failure_actual:
|
for m in get_failure_actual:
|
||||||
assert isinstance(m, FailureMessage) == True
|
assert isinstance(m, FailureMessage)
|
||||||
|
|
||||||
assert mocked_q_get_messages.call_count == 3
|
assert mocked_q_get_messages.call_count == 3
|
||||||
|
|
||||||
|
|
|
@ -2,7 +2,6 @@ import argparse
|
||||||
import os
|
import os
|
||||||
from subprocess import check_call
|
from subprocess import check_call
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from six import text_type
|
|
||||||
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
@ -38,7 +37,7 @@ def travis_build_package():
|
||||||
return failure
|
return failure
|
||||||
|
|
||||||
abs_dist_path = Path(os.environ["TRAVIS_BUILD_DIR"], "dist")
|
abs_dist_path = Path(os.environ["TRAVIS_BUILD_DIR"], "dist")
|
||||||
[create_package(package, text_type(abs_dist_path)) for package in package_list]
|
[create_package(package, str(abs_dist_path)) for package in package_list]
|
||||||
|
|
||||||
print("Produced:\n{}".format(list(abs_dist_path.glob("*"))))
|
print("Produced:\n{}".format(list(abs_dist_path.glob("*"))))
|
||||||
|
|
||||||
|
@ -53,9 +52,7 @@ def travis_build_package():
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
parser = argparse.ArgumentParser(description="Build Azure package.")
|
parser = argparse.ArgumentParser(description="Build Azure package.")
|
||||||
parser.add_argument("name", help="The package name")
|
parser.add_argument("name", help="The package name")
|
||||||
parser.add_argument(
|
parser.add_argument("--dest", "-d", default=DEFAULT_DESTINATION_FOLDER, help="Destination folder. Relative to the package dir. [default: %(default)s]")
|
||||||
"--dest", "-d", default=DEFAULT_DESTINATION_FOLDER, help="Destination folder. Relative to the package dir. [default: %(default)s]",
|
|
||||||
)
|
|
||||||
|
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
if args.name == "all":
|
if args.name == "all":
|
||||||
|
|
Загрузка…
Ссылка в новой задаче