-
Notifications
You must be signed in to change notification settings - Fork 21
/
example.py
executable file
·82 lines (60 loc) · 1.81 KB
/
example.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
76
77
78
79
80
81
82
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import sys
import alluxio
from alluxio import option
def colorize(code):
def _(text, bold=False):
c = code
if bold:
c = '1;%s' % c
return '\033[%sm%s\033[0m' % (c, text)
return _
green = colorize('32')
def info(s):
print(green(s))
def pretty_json(obj):
return json.dumps(obj, indent=2)
py_test_root_dir = '/py-test-dir'
py_test_nested_dir = '/py-test-dir/nested'
py_test = py_test_nested_dir + '/py-test'
py_test_renamed = py_test_root_dir + '/py-test-renamed'
client = alluxio.Client('localhost', 39999)
info("creating directory %s" % py_test_nested_dir)
opt = option.CreateDirectory(recursive=True)
client.create_directory(py_test_nested_dir, opt)
info("done")
info("writing to %s" % py_test)
with client.open(py_test, 'w') as f:
f.write('Alluxio works with Python!\n')
with open(sys.argv[0]) as this_file:
f.write(this_file)
info("done")
info("getting status of %s" % py_test)
stat = client.get_status(py_test)
print(pretty_json(stat.json()))
info("done")
info("renaming %s to %s" % (py_test, py_test_renamed))
client.rename(py_test, py_test_renamed)
info("done")
info("getting status of %s" % py_test_renamed)
stat = client.get_status(py_test_renamed)
print(pretty_json(stat.json()))
info("done")
info("reading %s" % py_test_renamed)
with client.open(py_test_renamed, 'r') as f:
print(f.read())
info("done")
info("listing status of paths under /")
root_stats = client.list_status('/')
for stat in root_stats:
print(pretty_json(stat.json()))
info("done")
info("deleting %s" % py_test_root_dir)
opt = option.Delete(recursive=True)
client.delete(py_test_root_dir, opt)
info("done")
info("asserting that %s is deleted" % py_test_root_dir)
assert not client.exists(py_test_root_dir)
info("done")