Commit 838272ed authored by 20after4's avatar 20after4
Browse files

Support resolving PHIDs automatically.

parent 2eb8cf02
import pprint
from enum import Enum
from collections import deque
from collections import UserDict, UserList, deque
from import MutableMapping, Iterable
import sqlite3
con = sqlite3.connect(':memory:')
......@@ -20,29 +21,29 @@ class DataIterator(object):
def __next__(self):
return Data(next(
class Data(object):
class Data(MutableMapping):
data = None
def __init__(self, data):
def __init__(self, data, parent=None):
# print(type(data)) = data
if parent:
self.parent = parent
def __getattr__(self, attr):
return self.__getitem__(attr)
def __getitem__(self, item):
itemdata =[item]
if isinstance(itemdata, list):
return DataList(itemdata)
elif isinstance(itemdata, dict):
return Data(itemdata)
if isinstance(itemdata, (dict, UserDict)):
return Data(itemdata, self)
elif isinstance(itemdata, (list, UserList, Iterable)):
return DataList(itemdata, self)
return itemdata
def __dir__(self):
return + self.__dict__.keys()
def __iter__(self):
return DataIterator(
......@@ -56,6 +57,7 @@ class Data(object):
def __repr__(self):
return pprint.pformat(, indent=2)
class DataList(Data):
def __getattr__(self, attr):
raise AttributeError()
......@@ -63,9 +65,11 @@ class DataList(Data):
def __dir__(self):
return dir(self.__dict__)
class Token(Enum):
ATTR = 1
ITEM = 2
class QueryBuilder(object):
......@@ -75,11 +79,15 @@ class QueryBuilder(object):
self.query = deque()
def __getitem__(self, key):
self.query.append((Token.ITEM, key))
return self
def __getattr__(self, key):
self.query.append((Token.ATTR, key))
return self
def parent(self):
from builtins import str
from collections import deque
from collections import UserDict, UserString, deque
from import Iterable
import json
import os
# todo: remove dependency on requests
import requests
from import DataIterator, Data
from ddd.phobjects import *
class Conduit(object):
phab_url = ''
......@@ -47,6 +50,29 @@ class Conduit(object):
return r
return ConduitResult(conduit=self, res=r, method=method, args=args)
def edit(self, method: str, objectidentifier: str, transactions: list):
Calls a conduit "edit" method which applies a list of transactions
to a specified object (specified by a phabricator identifier such as
T123, #project or a PHID of the appropriate type)
Raises an exception if something goes wrong and returns the decoded
conduit response otherwise.
data = {
"parameters": {
"transactions": transactions,
"objectidentifier": objectidentifier
data = flatten_for_post(args)
data['api.token'] = self.conduit_token
r ="{self.phab_url}{method}", data=data)
json = r.json()
if json['error_info'] is not None:
raise ConduitException(self.conduit, self, json['error_info'])
return json
class ConduitResult(object):
......@@ -58,13 +84,14 @@ class ConduitResult(object):
method = None
args = None
data = None
cursor = {}
cursor = None
def __init__(self, conduit: Conduit, res: requests.Response,
method: str, args: dict):
self.conduit = conduit
self.method = method
self.args = args
self.cursor = {}
def retry(self):
......@@ -103,6 +130,8 @@ class ConduitResult(object):
if "cursor" in self.result:
self.cursor = self.result['cursor']
self.cursor = {}
if "data" in self.result:
# Modern conduit methods return a result[data] and result{cursor}
......@@ -114,8 +143,36 @@ class ConduitResult(object): = self.result
def has_more(self):
after = self.cursor.get('after', None)
return after is not None
return self.cursor.get('after', None)
def resolve_phids(self, data=False):
if data is False:
data =
if isinstance(data, dict):
iter = data.items()
elif isinstance(data, (Iterable, list)) :
iter = enumerate(data)
for key, val in iter:
if isPHID(val):
data[key] = PHObject.instance(val)
elif isinstance(val, (list, dict)):
if data is
phids = [phid for phid in PHObject.instances.keys()]
args = {'phids': phids}
res = self.conduit.request(method='phid.query', args=args, raw=True)
objs = res.json()
for key, vals in objs['result'].items():
# for attr in vals.keys():
# setattr(PHObject.instances[key], attr, vals[attr])
return res
def __iter__(self):
return DataIterator(
......@@ -130,22 +187,13 @@ class ConduitResult(object):
return item in
class ConduitException(Exception):
def __init__(self, conduit: Conduit, result: ConduitResult, message: str):
self.conduit = conduit
self.result = result
self.message = message
def PHIDType(phid):
_, phidtype, phidhash = phid.split('-', 3)
return phidtype
def isPHID(value):
return isinstance(value, str) and str.startswith("PHID-")
def flatten_for_post(h, result=None, kk=None):
Since phab expects x-url-encoded form post data (meaning each
from enum import Enum
from typing import ClassVar
Phabricator Objects
PHObject and its subclasses are proxy objects that model
the relationships among objects in Phabricator's database.
Phabricator defines the relationships between objects by using PHIDs
which are long semi-structured strings that uniquely identify objects
and also carry simple type information about the object.
A PHID is made up of the string 'PHID' followed by several tokens
delimited by a '-', the first token indicates the type with 4 uppercase
ascii letters, the rest of the PHID is an arbitrary unique identifier string
Some known PHID types are enumerated in PHIDTypes class below.
class PHIDTypes(Enum):
PHOB = "PHObject"
PROJ = "Project"
TASK = "Task"
CMIT = "Commit"
XACT = "Transaction"
STRY = "FeedStory"
APPS = "Application"
PCOL = "ProjectColumn"
def isPHID(value):
return isinstance(value, str) and value.startswith("PHID-")
def PHIDType(phid):
parts = phid.split('-')
phidtype = parts[1]
if phidtype in PHIDTypes.__members__:
classname = PHIDTypes[phidtype].value
return classname
return phidtype
class PHObject(object):
PHObjects represent Phabricator objects such as Users and Tasks.
This class handles caching and insures that there is at most one instance
per unique phid.
id: int = 0
phid: str = ""
name: str = ""
dateCreated: int = 0
dateModified: int = 0
instances: ClassVar[dict] = {}
subclasses: ClassVar[dict] = {}
def __init_subclass__(cls, **kwargs):
__class__.subclasses[cls.__name__] = cls
def __init__(self, phid, **kwargs):
self.phid = phid
def __str__(self):
return self.fullName
def __repr__(self):
return f"<{self.__module__}.{self.__class__.__name__}{self.__dict__}>"
def update(self, data):
def loaded(self):
return len(self.__dict__.keys()) > 1
def instance(cls, phid):
if phid in __class__.instances:
return __class__.instances[phid]
phidtype = PHIDType(phid)
typecls = __class__.subclasses.get(phidtype, cls)
newinstance = typecls(phid)
__class__.instances[phid] = newinstance
return newinstance
class User(PHObject):
class Project(PHObject):
class ProjectColumn(PHObject):
class Task(PHObject):
class Transaction(PHObject):
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment