sketchingpy.local_data_struct

Implementation of data_struct for Sketch2DStatic and Sketch2DApp.

License:

BSD

 1"""Implementation of data_struct for Sketch2DStatic and Sketch2DApp.
 2
 3License:
 4    BSD
 5"""
 6
 7import csv
 8import io
 9import json
10import typing
11import urllib.request
12
13import sketchingpy.data_struct
14
15
16class LocalDataLayer(sketchingpy.data_struct.DataLayer):
17    """Implementation of DataLayer for desktop environment."""
18
19    def get_csv(self, path: str) -> sketchingpy.data_struct.Records:
20        if self._is_remote(path):
21            with urllib.request.urlopen(path) as response:
22                decoded = response.read().decode('utf-8')
23                decoded_io = io.StringIO(decoded)
24                reader = csv.DictReader(decoded_io)
25                rows = list(reader)
26        else:
27            with open(path) as f:
28                rows = list(csv.DictReader(f))
29
30        return rows
31
32    def write_csv(self, records: sketchingpy.data_struct.Records,
33        columns: sketchingpy.data_struct.Columns, path: str):
34        def build_record(target: typing.Dict) -> typing.Dict:
35            return dict(map(lambda key: (key, target[key]), columns))
36
37        records_serialized = map(build_record, records)
38
39        with open(path, 'w') as f:
40            writer = csv.DictWriter(f, fieldnames=columns)  # type: ignore
41            writer.writeheader()
42            writer.writerows(records_serialized)
43
44    def get_json(self, path: str):
45        if self._is_remote(path):
46            with urllib.request.urlopen(path) as response:
47                decoded = response.read().decode('utf-8')
48                target = json.loads(decoded)
49        else:
50            with open(path) as f:
51                target = json.load(f)
52
53        return target
54
55    def write_json(self, target, path: str):
56        with open(path, 'w') as f:
57            json.dump(target, f)
58
59    def get_text(self, path: str) -> str:
60        if self._is_remote(path):
61            with urllib.request.urlopen(path) as response:
62                text = response.read().decode('utf-8')
63        else:
64            with open(path) as f:
65                text = f.read()
66
67        return text
68
69    def write_text(self, target: str, path: str):
70        with open(path, 'w') as f:
71            f.write(target)
72
73    def _is_remote(self, path: str) -> bool:
74        return path.startswith('http://') or path.startswith('https://')
class LocalDataLayer(sketchingpy.data_struct.DataLayer): 
17class LocalDataLayer(sketchingpy.data_struct.DataLayer):
18    """Implementation of DataLayer for desktop environment."""
19
20    def get_csv(self, path: str) -> sketchingpy.data_struct.Records:
21        if self._is_remote(path):
22            with urllib.request.urlopen(path) as response:
23                decoded = response.read().decode('utf-8')
24                decoded_io = io.StringIO(decoded)
25                reader = csv.DictReader(decoded_io)
26                rows = list(reader)
27        else:
28            with open(path) as f:
29                rows = list(csv.DictReader(f))
30
31        return rows
32
33    def write_csv(self, records: sketchingpy.data_struct.Records,
34        columns: sketchingpy.data_struct.Columns, path: str):
35        def build_record(target: typing.Dict) -> typing.Dict:
36            return dict(map(lambda key: (key, target[key]), columns))
37
38        records_serialized = map(build_record, records)
39
40        with open(path, 'w') as f:
41            writer = csv.DictWriter(f, fieldnames=columns)  # type: ignore
42            writer.writeheader()
43            writer.writerows(records_serialized)
44
45    def get_json(self, path: str):
46        if self._is_remote(path):
47            with urllib.request.urlopen(path) as response:
48                decoded = response.read().decode('utf-8')
49                target = json.loads(decoded)
50        else:
51            with open(path) as f:
52                target = json.load(f)
53
54        return target
55
56    def write_json(self, target, path: str):
57        with open(path, 'w') as f:
58            json.dump(target, f)
59
60    def get_text(self, path: str) -> str:
61        if self._is_remote(path):
62            with urllib.request.urlopen(path) as response:
63                text = response.read().decode('utf-8')
64        else:
65            with open(path) as f:
66                text = f.read()
67
68        return text
69
70    def write_text(self, target: str, path: str):
71        with open(path, 'w') as f:
72            f.write(target)
73
74    def _is_remote(self, path: str) -> bool:
75        return path.startswith('http://') or path.startswith('https://')

Implementation of DataLayer for desktop environment.

def get_csv(self, path: str) -> Iterable[Dict]: 
20    def get_csv(self, path: str) -> sketchingpy.data_struct.Records:
21        if self._is_remote(path):
22            with urllib.request.urlopen(path) as response:
23                decoded = response.read().decode('utf-8')
24                decoded_io = io.StringIO(decoded)
25                reader = csv.DictReader(decoded_io)
26                rows = list(reader)
27        else:
28            with open(path) as f:
29                rows = list(csv.DictReader(f))
30
31        return rows

Load a CSV file as a list of dictionaries.

Load a CSV file and parse it as a list where each row after the "header row" becomes a dictionary.

Arguments:
  • path: The location at which the CSV file can be found.
Returns:

List of dictionary.

def write_csv(self, records: Iterable[Dict], columns: Iterable[str], path: str): 
33    def write_csv(self, records: sketchingpy.data_struct.Records,
34        columns: sketchingpy.data_struct.Columns, path: str):
35        def build_record(target: typing.Dict) -> typing.Dict:
36            return dict(map(lambda key: (key, target[key]), columns))
37
38        records_serialized = map(build_record, records)
39
40        with open(path, 'w') as f:
41            writer = csv.DictWriter(f, fieldnames=columns)  # type: ignore
42            writer.writeheader()
43            writer.writerows(records_serialized)

Write a list of dictionaries as a CSV file.

Write a CSV file with header row, saving it to local file system or offering it as a download in the browser.

Arguments:
  • records: List of dictionaries to be written.
  • columns: Ordered list of columns to include in the CSV file.
  • path: The location at which the file should be written.
def get_json(self, path: str): 
45    def get_json(self, path: str):
46        if self._is_remote(path):
47            with urllib.request.urlopen(path) as response:
48                decoded = response.read().decode('utf-8')
49                target = json.loads(decoded)
50        else:
51            with open(path) as f:
52                target = json.load(f)
53
54        return target

Read a JSON file.

Read a JSON file either from local file system or the network.

Arguments:
  • path: The location at which the JSON file can be found.
Returns:

Loaded JSON content.

def write_json(self, target, path: str): 
56    def write_json(self, target, path: str):
57        with open(path, 'w') as f:
58            json.dump(target, f)

Write a JSON file.

Write a JSON file, saving it to local file system or offering it as a download in the browser.

Arguments:
  • target: The value to write as JSON.
  • path: The location at which the file should be written.
def get_text(self, path: str) -> str: 
60    def get_text(self, path: str) -> str:
61        if self._is_remote(path):
62            with urllib.request.urlopen(path) as response:
63                text = response.read().decode('utf-8')
64        else:
65            with open(path) as f:
66                text = f.read()
67
68        return text

Read a text file.

Read a text file either from local file system or the network.

Arguments:
  • path: The location where the file can be found.
Returns:

Loaded content as a string.

def write_text(self, target: str, path: str): 
70    def write_text(self, target: str, path: str):
71        with open(path, 'w') as f:
72            f.write(target)

Write a text file.

Write a text file, saving it to local file system or offering it as a download in the browser.

Arguments:
  • target: The contents to write.
  • path: The location at which the file should be written.