-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
75 lines (63 loc) · 1.94 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import os
from dotenv import load_dotenv
import json
import random
import string
import requests
# Load environment variables
dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
load_dotenv(dotenv_path)
options = {}
for name in [
'USERNAME',
'PASSWORD',
'HOSTNAME'
]:
options[name] = os.environ.get(name)
hostname = options['HOSTNAME'].rstrip('/') # remove trailing slash
auth = (options['USERNAME'], options['PASSWORD'])
# Create dataset
def create_dataset(metadata):
r = requests.post(hostname + '/api/views.json',
auth=auth,
data=json.dumps(metadata)
)
return r.json()
# Make dataset public
def make_dataset_public(dataset_id):
params = {
'method': 'setPermission',
'value': 'public.read'
}
r = requests.put(hostname + '/api/views/' + dataset_id, auth=auth, params=params)
return r.status_code == 200
# Publish dataset
def publish_dataset(dataset):
r = requests.post(hostname + '/api/views/' + dataset_id + '/publication.json', auth=auth)
return r.status_code == 200
# Create a sample dataset
if __name__ == '__main__':
# Load fields from sample file
with open('sample_fields.json') as sample_fields_file:
sample_fields = json.load(sample_fields_file)
payload = {
'name': 'Test ' + ''.join(random.choice(string.lowercase) for i in range(6)), # random string
'description': 'Created via socrata-dataset-creator for testing purposes',
'columns': sample_fields
}
# I'm probably doing try/catch incorrectly. Any suggestions? (via issue or PR)
try:
dataset = create_dataset(payload)
dataset_id = dataset[u'id']
print('Dataset created at %s/d/%s' % (hostname, dataset_id))
if make_dataset_public(dataset_id):
print('Made %s public' % dataset_id)
else:
print('Error making %s public' % dataset_id)
if publish_dataset(dataset_id):
print('Published %s' % dataset_id)
else:
print('Error publishing %s' % dataset_id)
except Error as e:
print('An error occurred: %s - %s' % (e.__class__, e))
raise