oerppy-hackers team mailing list archive
-
oerppy-hackers team
-
Mailing list archive
-
Message #00014
[Branch ~oerppy-hackers/oerppy/trunk] Rev 34: * Started creating object data models for results (including unit tests).
------------------------------------------------------------
revno: 34
committer: duncan@xxxxxxxxxx
branch nick: trunk
timestamp: Thu 2011-06-02 16:46:11 -0600
message:
* Started creating object data models for results (including unit tests).
* Converted two of the main query methods in the Canonical addon to use the
new data model classes; this has resulted in the addon code being much more
readable.
* Broke remaining method into lots of smaller methods that will now be much
easer to convert into data-model-based queries.
* Minor tweaks to the Makefile.
added:
oerppy/tests/test_model.py
modified:
Makefile
oerppy/addons/canonical/query.py
oerppy/client.py
oerppy/model.py
--
lp:oerppy
https://code.launchpad.net/~oerppy-hackers/oerppy/trunk
Your team oerppy Hackers is subscribed to branch lp:oerppy.
To unsubscribe from this branch go to https://code.launchpad.net/~oerppy-hackers/oerppy/trunk/+edit-subscription
=== modified file 'Makefile'
--- Makefile 2011-06-02 19:03:01 +0000
+++ Makefile 2011-06-02 22:46:11 +0000
@@ -3,6 +3,7 @@
find ./ -name "*.pyc" -exec rm {} \;
find ./ -name "*.pyo" -exec rm {} \;
find . -name "*.sw[op]" -exec rm {} \;
+ rm -rf MSG.backup _trial_temp/
check:
trial oerppy
=== modified file 'oerppy/addons/canonical/query.py'
--- oerppy/addons/canonical/query.py 2011-06-02 19:03:01 +0000
+++ oerppy/addons/canonical/query.py 2011-06-02 22:46:11 +0000
@@ -1,5 +1,6 @@
from datetime import date, timedelta
+from oerppy import model
from oerppy.addons.canonical import const
@@ -26,7 +27,7 @@
if not domains:
domains = []
if not self.user_dept:
- self.get_departments()
+ self.set_departments()
if report == const.TIMESHEET_HOURS:
records = self.timesheet_hours(domains)
elif report == const.TIMESHEETS_WEEKLY:
@@ -35,95 +36,89 @@
raise OpenERPPyParameterError("Unknown value for 'report.'")
return records
- # XXX many of these methods will become more simplified through the use of
- # a parser object... cleanly dividing code along the lines of
- # functionality:
- # 1) create a query and send it out
- # 2) parse the result set
- # 3) convert the parsed data into a model instance
- # then, of course, one is able to work with the data quite easily
-
- def get_departments(self):
+ def set_departments(self):
"""
Build a lookup dictionary of department vs user_id.
"""
- employees = self.client.searchfields(
- "hr.employee",
- [],
- ['name','department_id','resource_id','user_id'])
-
- # Build dictionary of resource_id vs department
- user_dept = {}
- for empl in employees:
- if empl['department_id'] and empl['user_id']:
- user_dept[empl['user_id'][0]] = empl['department_id'][1]
- else:
- if empl['user_id']:
- user_dept[empl['user_id'][0]] = '?'
- else:
- user_dept[0] = '?'
-
- self.user_dept = user_dept
-
- def timesheet_hours(self, domain):
- recs = self.client.searchfields(
- "project.task.work",
- domain,
- self.fields)
- records = []
- for r in recs:
- line = {
- 'Project': r['project_id'][1],
- 'Task': r['task_id'][1],
- 'User': r['user_id'][1],
- 'Hours': r['hours'],
- 'Department': self.user_dept.get(r['user_id'][0], '?'),
- 'Date': r['date'],
- 'Week': r['week_start'],
- 'Year': r['date'][0:4],
- 'Month': r['date'][5:7],
- }
- records.append(line)
- return records
-
- def weekly_timesheets(self, domain):
- """
- Get the weekly timeentry records, highlighting the missing weeks.
- """
- # Get user Ids of timesheet users and the weeks we're interested in
+ raw_results = self.client.searchfields(
+ entity="hr.employee",
+ query=[],
+ fields=['name','department_id','resource_id','user_id'])
+ results = model.HREmployeeSet(raw_results)
+ # build dictionary of resource_id vs department
+ self.user_dept = results.build_department_lookup_table()
+
+ def timesheet_hours(self, domains):
+ raw_results = self.client.searchfields(
+ entity="project.task.work",
+ query=domains,
+ fields=self.fields)
+ return model.ProjectTaskWorkSet(raw_results, self.user_dept)
+
+ def get_all_timeentry_weeks(self):
+ """
+ Get timeentry data since we've been tracking it.
+ """
today = date.today()
last_week = today - timedelta(days=-7)
weeks = self.client.searchfields(
- 'project.timeentry.week',
- [
+ entity='project.timeentry.week',
+ query=[
(['week','>','2010-12-31']),
('week','<=', last_week.isoformat())
],
+ fields=[])
+
+ def get_timeentry(self, week):
+ """
+ Get the timeentry for a given week.
+ """
+ timesheets = self.client.searchfields(
+ 'project.timeentry',
+ [('week_start','=',week['week']),('user_id','in', user_ids)],
[])
+ return timesheets
+
+ def get_oem_user_ids(self):
+ """
+ Get the IDs for everyone who uses the OEM timesheets.
+ """
user_fields = self.client.searchfields(
'canonical.user.category',
[('code','=','OEMTIMESHEET')],
['user_ids'])
- user_ids = user_fields[0]['user_ids']
+ return user_fields[0]['user_ids']
+
+ def get_user_names(self, user_ids):
+ """
+ Get the OpenERP user names for the user IDs provided.
+ """
users = self.client.searchfields(
'res.users' ,
[('id','in', user_ids)],
['id', 'name'])
-
- # Build list of names for each user id
+ return users
+
+ def weekly_timesheets(self, domains):
+ """
+ Get the weekly timeentry records, highlighting the missing weeks.
+ """
+ # get user IDs of timesheet users and the weeks we're interested in
+
+ weeks = self.get_all_timeentry_weeks()
+ user_ids = self.get_oem_user_ids()
+ users = self.get_users_names(user_ids)
+ # build list of names for each user id
user_list = {}
for u in users:
user_list[u['id']] = u['name']
records = []
- for w in weeks:
- # Get timesheets for the week
- timesheets = self.client.searchfields(
- 'project.timeentry',
- [('week_start','=',w['week']),('user_id','in', user_ids)],
- [])
+ for week in weeks:
+ # get timesheets for the week
+ timesheets = self.get_timeentry(week["week"])
- # Store lookup of total time by user for this week
+ # store lookup of total time by user for this week
total_time = {}
if timesheets:
for t in timesheets:
@@ -132,7 +127,7 @@
for u in user_ids:
if not total_time.get(u,None):
- # Missing timesheet
+ # missing timesheet
line = {
'Week Commencing': w['week'],
'User': user_list[u],
=== modified file 'oerppy/client.py'
--- oerppy/client.py 2011-06-02 19:03:01 +0000
+++ oerppy/client.py 2011-06-02 22:46:11 +0000
@@ -69,7 +69,7 @@
return self.endpoints.object.execute(entity, "read", ids, fields)
@parse_results
- def searchfields(self, entity, query, fields):
+ def searchfields(self, entity="", query="", fields=None):
"""
"""
ids = self.search(entity, query)
=== modified file 'oerppy/model.py'
--- oerppy/model.py 2011-06-01 04:50:04 +0000
+++ oerppy/model.py 2011-06-02 22:46:11 +0000
@@ -1,4 +1,82 @@
+
class Model(object):
"""
Base class for representing result sets and result data.
"""
+
+
+class ResultSet(set):
+ """
+ Base class for collections of models, all together representing result sets
+ in queries to the OpenERP server.
+ """
+ model = Model
+
+ def __init__(self, results, *args, **kwds):
+ for row in results:
+ self.add(self.model(row, *args, **kwds))
+
+
+class HREmployee(Model):
+ """
+ This is an object model that maps to search results from OpenERP's
+ "hr.employee" entity.
+ """
+ def __init__(self, row):
+ self.name = row["name"][0]
+ self.department_id = row["department_id"][1]
+ self.resource_id = row["resource_id"][0]
+ self.user_id = row["user_id"][0]
+
+
+class HREmployeeSet(ResultSet):
+ """
+ An object for collecting HREmployees together.
+ """
+ model = HREmployee
+
+ def get_user(self, user_id):
+ for user in self:
+ if user.user_id == user_id:
+ return user
+
+ def build_department_lookup_table(self):
+ table = {}
+ table["unknown"] = 0
+ for user in self:
+ if user.department_id and user.user_id:
+ table[user.user_id] = user.department_id
+ elif user.user_id:
+ table[user.user_id] = "?"
+ else:
+ table["unknown"] += 1
+ return table
+
+
+class ProjectTaskWork(Model):
+ """
+ This is an object model that maps to search results from OpenERP's
+ "project.task.work" entity.
+ """
+ def __init__(self, row, depts):
+ self.project_id = row['project_id'][1]
+ self.task_id = row['task_id'][1]
+ self.user_id = row['user_id'][1]
+ self.department_id = depts.get(self.user_id, "?")
+ self.hours = row['hours']
+ self.date = row['date']
+ self.year = int(row['date'][0:4])
+ self.month = int(row['date'][5:7])
+ self.week = row['week_start']
+
+
+class ProjectTaskWorkSet(ResultSet):
+ """
+ An object for collecting HREmployees together.
+ """
+ model = ProjectTaskWork
+
+ def get_work(self, project_id):
+ for project in self:
+ if project.project_id == project_id:
+ return project
=== added file 'oerppy/tests/test_model.py'
--- oerppy/tests/test_model.py 1970-01-01 00:00:00 +0000
+++ oerppy/tests/test_model.py 2011-06-02 22:46:11 +0000
@@ -0,0 +1,93 @@
+import unittest
+
+from oerppy import model
+
+
+class HRTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.user_results = [
+ {"name": ["Bob"],
+ "user_id": [314],
+ "department_id": ["what's this field?", 1],
+ "resource_id": [34]},
+ {"name": ["Alice"],
+ "user_id": [315],
+ "department_id": ["UKN", 2],
+ "resource_id": [37]},
+ {"name": ["Carol"],
+ "user_id": [316],
+ "department_id": ["UKN", None],
+ "resource_id": [42]},
+ ]
+
+ def test_hr_employee(self):
+ employee = model.HREmployee(self.user_results[1])
+ self.assertEqual(employee.user_id, 315)
+ self.assertEqual(employee.name, "Alice")
+ self.assertEqual(employee.department_id, 2)
+ self.assertEqual(employee.resource_id, 37)
+
+ def test_hr_employee_set(self):
+ results = model.HREmployeeSet(self.user_results)
+ self.assertEqual(len(results), 3)
+ user = results.get_user(314)
+ self.assertEqual(user.name, "Bob")
+ table = results.build_department_lookup_table()
+ self.assertEqual(table.get(315), 2)
+ self.assertEqual(table.get(316), "?")
+
+class ProjectTaskWorkTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.user_results = [
+ {"name": ["Bob"],
+ "user_id": [314],
+ "department_id": ["what's this field?", 1],
+ "resource_id": [34]},
+ {"name": ["Alice"],
+ "user_id": [315],
+ "department_id": ["UKN", 2],
+ "resource_id": [37]},
+ {"name": ["Carol"],
+ "user_id": [316],
+ "department_id": ["UKN", None],
+ "resource_id": [42]},
+ ]
+ self.work_results = [
+ {"project_id": ["UKN", 1],
+ "task_id": ["UKN", 1001],
+ "user_id": ["UKN", 314],
+ "hours": 273,
+ "department_id": ["what's this field?", 1],
+ "date": "2011-05-01",
+ "week_start": 172},
+ {"project_id": ["UKN", 2],
+ "task_id": ["UKN", 1002],
+ "user_id": ["UKN", 315],
+ "hours": 735,
+ "department_id": ["what's this field?", 2],
+ "date": "2011-05-01",
+ "week_start": 172},
+ ]
+ self.results = model.HREmployeeSet(self.user_results)
+ self.table = self.results.build_department_lookup_table()
+
+ def test_project_task_work(self):
+
+ work = model.ProjectTaskWork(self.work_results[1], self.table)
+ self.assertEqual(work.project_id, 2)
+ self.assertEqual(work.task_id, 1002)
+ self.assertEqual(work.user_id, 315)
+ self.assertEqual(work.department_id, 2)
+ self.assertEqual(work.hours, 735)
+ self.assertEqual(work.date, "2011-05-01")
+ self.assertEqual(work.year, 2011)
+ self.assertEqual(work.month, 5)
+ self.assertEqual(work.week, 172)
+
+ def test_project_task_work_set(self):
+ results = model.ProjectTaskWorkSet(self.work_results, self.table)
+ self.assertEqual(len(results), 2)
+ work = results.get_work(1)
+ self.assertEqual(work.hours, 273)
