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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
from irods.api_number import api_number
from irods.message import iRODSMessage, TicketAdminRequest
from irods.models import TicketQuery
import random
import string
import logging
import datetime
import calendar
logger = logging.getLogger(__name__)
def get_epoch_seconds(utc_timestamp):
epoch = None
try:
epoch = int(utc_timestamp)
except ValueError:
pass
if epoch is not None:
return epoch
HUMAN_READABLE_DATE = "%Y-%m-%d.%H:%M:%S"
try:
x = datetime.datetime.strptime(utc_timestamp, HUMAN_READABLE_DATE)
return calendar.timegm(x.timetuple())
except ValueError:
raise # final try at conversion, so a failure is an error
class Ticket:
def __init__(self, session, ticket="", result=None, allow_punctuation=False):
self._session = session
try:
if result is not None:
ticket = result[TicketQuery.Ticket.string]
except TypeError:
raise RuntimeError(
"If specified, 'result' parameter must be a TicketQuery.Ticket search result"
)
self._ticket = (
ticket if ticket else self._generate(allow_punctuation=allow_punctuation)
)
@property
def session(self):
return self._session
@property
def ticket(self):
"""Return the unique string associated with the ticket object."""
return self._ticket
# Provide 'string' property such that self.string is a synonym for self.ticket
string = ticket
def _generate(self, length=15, allow_punctuation=False):
source_characters = string.ascii_letters + string.digits
if allow_punctuation:
source_characters += string.punctuation
return "".join(
random.SystemRandom().choice(source_characters) for _ in range(length)
)
def _api_request(self, cmd_string, *args, **opts):
with self.session.pool.get_connection() as conn:
self._lowlevel_api_request(conn, cmd_string, self.ticket, *args, **opts)
return self
@staticmethod
def _lowlevel_api_request(conn_, cmd_string, ticket_string, *args, **opts):
message_body = TicketAdminRequest(cmd_string, ticket_string, *args, **opts)
message = iRODSMessage(
"RODS_API_REQ", msg=message_body, int_info=api_number["TICKET_ADMIN_AN"]
)
conn_.send(message)
response = conn_.recv()
return response
def issue(self, permission, target, **opt):
return self._api_request("create", permission, target, **opt)
create = issue
def modify(self, *args, **opt):
arglist = list(args)
if arglist[0].lower().startswith("expir"):
arglist[1] = str(get_epoch_seconds(utc_timestamp=arglist[1]))
return self._api_request("mod", *arglist, **opt)
def supply(self, **opt):
self.session.ticket__ = self._ticket
return self
def delete(self, **opt):
"""
Delete the iRODS ticket.
This applies to a Ticket object on which issue() has been called or, as the case may
be, to a Ticket initialized with a ticket string already existing in the object catalog.
The deleted object is returned, but may not be used further except for local purposes
such as extracting the string. E.g.
for t in tickets:
print(t.delete().string, "being deleted")
"""
return self._api_request("delete", **opt)
|
