Skip to content
GitLab
Commit 73c0174e authored by Roland Koller's avatar Roland Koller
Browse files

Support typed portfolios

parent 2e07765c
Hide whitespace changes
Inline Side-by-side
ecmind_blue_client_portfolio/portfolio.py
View file @ 73c0174e
......@@ -8,7 +8,7 @@ from datetime import datetime
@dataclass(frozen=True)
class PortfolioObject:
"""Class to store portfolio member objects."""
id: int
id: int
type_id: int
......@@ -20,11 +20,14 @@ class Portfolio:
creator: str
recipient: str
subject: str
type_id: Union[None, int]
objects: List[PortfolioObject]
def search(client:Client, id:int=None, creator:str=None, recipient:str=None, subject:str=None, created:datetime=None, created_max_date:datetime=None, garbage_mode:bool=False) -> List[Portfolio]:
def search(client: Client, id: int = None, creator: str = None, recipient: str = None,
subject: str = None, created: datetime = None, created_max_date: datetime = None,
garbage_mode: bool = False, type_id: Union[None, int] = None
) -> List[Portfolio]:
"""Search for portfolios.
Keyword arguments:
......@@ -34,6 +37,7 @@ def search(client:Client, id:int=None, creator:str=None, recipient:str=None, sub
recipient -- (Optional) Username of the portfolio recipient to search for.
subject -- (Optional) Subject of the portfolio to search for.
garbage_mode -- (Optional) bool indicating, if the query searches the recycle bin instead of non-deleted portfolios, default = False.
type_id -- (Optional) object type if for typed portfolios.
"""
portfolio_xml = XmlElement.from_object('Portfolio', {
......@@ -47,16 +51,21 @@ def search(client:Client, id:int=None, creator:str=None, recipient:str=None, sub
if id != None:
portfolio_xml.set('id', str(id))
if type_id != None:
portfolio_xml.set('objtype', str(type_id))
job = Job(
'dms.RetrievePortfolios',
'dms.RetrievePortfolios',
Flags=0,
GarbageMode=1 if garbage_mode else 0,
Created_to=int(created_max_date.strftime('%s')) if created_max_date else 0
Created_to=int(created_max_date.strftime('%s')
) if created_max_date else 0
)
job.append(Param('PortfolioXML', ParamTypes.BASE64, portfolio_xml.to_string()))
job.append(Param('PortfolioXML', ParamTypes.BASE64,
portfolio_xml.to_string()))
result = client.execute(job)
if result.return_code != 0:
......@@ -79,23 +88,28 @@ def search(client:Client, id:int=None, creator:str=None, recipient:str=None, sub
creator=portfolio['@creator'],
recipient=portfolio['@recipient'],
subject=portfolio['@subject'],
objects=[ PortfolioObject(id=obj['@id'], type_id=obj['@objecttype_id']) for obj in portfolio['Objects']['Object'] ] if (portfolio['Objects'] and 'Object' in portfolio['Objects']) else []
type_id=(int(portfolio['@objtype']) if '@objtype' in portfolio and int(portfolio['@objtype']) > 0 else None),
objects=[PortfolioObject(id=obj['@id'], type_id=obj['@objecttype_id']) for obj in portfolio['Objects']
['Object']] if (portfolio['Objects'] and 'Object' in portfolio['Objects']) else []
))
return result
def user_favorites(client:Client, username:str) -> Union[Portfolio, None]:
def user_favorites(client: Client, username: str) -> Union[Portfolio, None]:
"""Shortcut function to search for a users favorites portfolio."""
p = search(client, creator=username, recipient=username, created=datetime(1970, 1, 1, 1, 0, 3))
p = search(client, creator=username, recipient=username,
created=datetime(1970, 1, 1, 1, 0, 3))
if len(p) == 1:
return p[0]
else:
return None
def create(client:Client, creator:str=None, recipient:str=None, subject:str=None, objects:Optional[List[PortfolioObject]]=None) -> int:
def create(client: Client, creator: str = None, recipient: str = None,
subject: str = None, objects: Optional[List[PortfolioObject]] = None,
type_id: Union[None, int] = None) -> int:
"""Create a portfolio and return its id.
Keyword arguments:
......@@ -104,10 +118,11 @@ def create(client:Client, creator:str=None, recipient:str=None, subject:str=None
recipient -- (Optional) Username of the portfolio recipient to search for.
subject -- (Optional) Subject of the portfolio to search for.
objects -- (Optional) List of `PortfolioObject`s.
"""
type_id -- (Optional) object type if for typed portfolios.
"""
if objects:
portfolio_objects = {'Object': [
{'@id': obj.id, '@objecttype_id': obj.type_id}
portfolio_objects = {'Object': [
{'@id': obj.id, '@objecttype_id': obj.type_id}
for obj in objects
]}
else:
......@@ -120,22 +135,26 @@ def create(client:Client, creator:str=None, recipient:str=None, subject:str=None
'Objects': portfolio_objects
})
if type_id != None:
portfolio_xml.set('objtype', str(type_id))
job = Job(
jobname='dms.AddPortfolio',
Flags=0,
Mode=1
)
job.append(Param('PortfolioXML', ParamTypes.BASE64, portfolio_xml.to_string()))
job.append(Param('PortfolioXML', ParamTypes.BASE64,
portfolio_xml.to_string()))
result = client.execute(job)
if result.return_code != 0:
raise RuntimeError(result.error_message)
return result.values['PortfolioID']
def delete(client:Client, portfolio_or_id:Union[Portfolio, int]) -> None:
def delete(client: Client, portfolio_or_id: Union[Portfolio, int]) -> None:
'''Delete a portfolio by its id.'''
if isinstance(portfolio_or_id, Portfolio):
id = portfolio_or_id.id
......@@ -145,18 +164,19 @@ def delete(client:Client, portfolio_or_id:Union[Portfolio, int]) -> None:
portfolio_xml = XmlElement.from_object('Portfolio', {
'@id': id
})
job = Job(jobname='dms.DelPortfolio', Flags=0)
job.append(Param('PortfolioXML', ParamTypes.BASE64, portfolio_xml.to_string()))
job.append(Param('PortfolioXML', ParamTypes.BASE64,
portfolio_xml.to_string()))
result = client.execute(job)
if result.return_code != 0:
raise RuntimeError(result.error_message)
return None
def delete_for(client:Client, recipient:str, subject:str) -> None:
def delete_for(client: Client, recipient: str, subject: str) -> None:
'''Delete a portfolio by its recipient and subject.'''
portfolio_xml = XmlElement.from_object('Portfolio', {
......@@ -165,16 +185,17 @@ def delete_for(client:Client, recipient:str, subject:str) -> None:
})
job = Job(jobname='dms.DelPortfolio', Flags=0)
job.append(Param('PortfolioXML', ParamTypes.BASE64, portfolio_xml.to_string()))
job.append(Param('PortfolioXML', ParamTypes.BASE64,
portfolio_xml.to_string()))
result = client.execute(job)
if result.return_code != 0:
raise RuntimeError(result.error_message)
return None
def remove_objects(client:Client, portfolio_or_id:Union[Portfolio, int], objects:List[PortfolioObject]):
def remove_objects(client: Client, portfolio_or_id: Union[Portfolio, int], objects: List[PortfolioObject]):
'''Remove objects from a portfolio.'''
if isinstance(portfolio_or_id, Portfolio):
......@@ -184,23 +205,24 @@ def remove_objects(client:Client, portfolio_or_id:Union[Portfolio, int], objects
portfolio_xml = XmlElement.from_object('Portfolio', {
'@id': id,
'Objects': {'Object': [
{'@id': obj.id, '@objecttype_id': obj.type_id}
'Objects': {'Object': [
{'@id': obj.id, '@objecttype_id': obj.type_id}
for obj in objects
]}
})
job = Job(jobname='dms.RemoveFromportfolio', Flags=0)
job.append(Param('PortfolioXML', ParamTypes.BASE64, portfolio_xml.to_string()))
job.append(Param('PortfolioXML', ParamTypes.BASE64,
portfolio_xml.to_string()))
result = client.execute(job)
if result.return_code != 0:
raise RuntimeError(result.error_message)
return None
def add_objects(client:Client, portfolio_or_id:Union[Portfolio, int], objects:List[PortfolioObject], replace_existing_object_list:bool=False):
def add_objects(client: Client, portfolio_or_id: Union[Portfolio, int], objects: List[PortfolioObject], replace_existing_object_list: bool = False):
'''Add new objects to a portfolio. Set `replace_existing_object_list` to clear portfolio upfront.'''
if isinstance(portfolio_or_id, Portfolio):
......@@ -210,28 +232,28 @@ def add_objects(client:Client, portfolio_or_id:Union[Portfolio, int], objects:Li
portfolio_xml = XmlElement.from_object('Portfolio', {
'@id': id,
'Objects': {'Object': [
{'@id': obj.id, '@objecttype_id': obj.type_id}
'Objects': {'Object': [
{'@id': obj.id, '@objecttype_id': obj.type_id}
for obj in objects
]}
})
job = Job(
jobname='dms.ModPortfolio',
Flags=0,
jobname='dms.ModPortfolio',
Flags=0,
Mode=(1 if replace_existing_object_list else 0)
)
job.append(Param('PortfolioXML', ParamTypes.BASE64, portfolio_xml.to_string()))
job.append(Param('PortfolioXML', ParamTypes.BASE64,
portfolio_xml.to_string()))
result = client.execute(job)
if result.return_code != 0:
raise RuntimeError(result.error_message)
return None
return None
def clear_objects(client:Client, portfolio_or_id:Union[Portfolio, int]):
def clear_objects(client: Client, portfolio_or_id: Union[Portfolio, int]):
'''Shortcut function to remove all objects from a portfolio.'''
return add_objects(client, portfolio_or_id, [], True)
\ No newline at end of file
return add_objects(client, portfolio_or_id, [], True)
setup.py
View file @ 73c0174e
......@@ -5,7 +5,7 @@ with open('README.md', 'r') as fh:
setuptools.setup(
name='ecmind_blue_client_portfolio',
version='0.0.2',
version='0.0.3',
author='Roland Koller',
author_email='info@ecmind.ch',
description='Helper modules for the ecmind_blue_client to ease the work with the portfolio API functions.',
......
tests.py
View file @ 73c0174e
......@@ -96,5 +96,23 @@ class TestMethods(unittest.TestCase):
portfolio.delete(self.client, created_id)
def test_typed_portfolio(self):
type_id = 262144
created_id = portfolio.create(self.client, 'ROOT', 'ROOT', 'UnitTest7', [PortfolioObject(38, type_id)], type_id=type_id)
p1 = portfolio.search(self.client, created_id, type_id=type_id)[0]
self.assertEqual(p1.type_id, type_id)
self.assertEqual(len(p1.objects), 1)
with self.assertRaises(RuntimeError):
portfolio.add_objects(self.client, p1, [PortfolioObject(301, 262150)])
portfolio.add_objects(self.client, p1, [PortfolioObject(43, type_id)])
p2 = portfolio.search(self.client, created_id)[0]
self.assertEqual(len(p2.objects), 2)
portfolio.delete(self.client, created_id)
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
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