Skip to content

Instantly share code, notes, and snippets.

@punchagan

punchagan/backend_fogbugz.py

Created January 16, 2014 15:02
Show Gist options
  • Save punchagan/8456363 to your computer and use it in GitHub Desktop.
Save punchagan/8456363 to your computer and use it in GitHub Desktop.
Download ZIP
GTG Backend for FogBugz
Raw
backend_fogbugz.py
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Getting Things GNOME! - a personal organizer for the GNOME desktop
# Copyright (c) 2008-2013 - Lionel Dricot & Bertrand Rousseau
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program. If not, see <http://www.gnu.org/licenses/>.
# -----------------------------------------------------------------------------
from datetime import datetime
import os
import uuid
from GTG import _
from GTG.backends.genericbackend import GenericBackend
from GTG.backends.backendsignals import BackendSignals
from GTG.backends.periodicimportbackend import PeriodicImportBackend
from GTG.backends.syncengine import SyncEngine, SyncMeme
from GTG.tools.logger import Log
from GTG.core.task import Task
from fogbugz import FogBugz, FogBugzAPIError
""" Backend for importing fogbugz issues in GTG
Dependencies:
* fogbugz
"""
class Backend(PeriodicImportBackend):
_general_description = {
GenericBackend.BACKEND_NAME: "backend_fogbugz",
GenericBackend.BACKEND_HUMAN_NAME: _("FogBugz"),
GenericBackend.BACKEND_AUTHORS: ['Authors who?'],
GenericBackend.BACKEND_TYPE: GenericBackend.TYPE_READONLY,
GenericBackend.BACKEND_DESCRIPTION:
_(
"This synchronization service lets you import the issues found"
" on FogBugz."
#fixme: what do we do?
"Please note that this is a read only synchronization service."
#fixme: what does it mean?
),
}
_static_parameters = {
"period": {
GenericBackend.PARAM_TYPE: GenericBackend.TYPE_INT,
GenericBackend.PARAM_DEFAULT_VALUE: 1,
},
"username": {
GenericBackend.PARAM_TYPE: GenericBackend.TYPE_STRING,
GenericBackend.PARAM_DEFAULT_VALUE: 'insert your email-id', },
"password": {
GenericBackend.PARAM_TYPE: GenericBackend.TYPE_PASSWORD,
GenericBackend.PARAM_DEFAULT_VALUE: '', },
"service-url": {
GenericBackend.PARAM_TYPE: GenericBackend.TYPE_STRING,
GenericBackend.PARAM_DEFAULT_VALUE: 'https://x.fogbugz.com/',
},
"project-id": {
GenericBackend.PARAM_TYPE: GenericBackend.TYPE_STRING,
GenericBackend.PARAM_DEFAULT_VALUE: ''
},
"tag-with-project-name": {
GenericBackend.PARAM_TYPE: GenericBackend.TYPE_BOOL,
GenericBackend.PARAM_DEFAULT_VALUE: True
},
}
def __init__(self, parameters):
""" See GenericBackend for an explanation of this function.
Re-loads the saved state of the synchronization.
"""
super(Backend, self).__init__(parameters)
# loading the saved state of the synchronization, if any
self.data_path = os.path.join(
'backends/fogbugz/', 'sync_engine-' + self.get_id()
)
self.sync_engine = self._load_pickled_file(
self.data_path, SyncEngine()
)
self._fb= None
def save_state(self):
""" Saves the state of the synchronization. """
self._store_pickled_file(self.data_path, self.sync_engine)
def do_periodic_import(self):
""" Import the issues in fogbugz, periodically. """
self._login()
fb = self._fb
if fb is None:
return
project_id = self._parameters.get('project-id', '')
milestones = fb.listFixFors(ixProject=int(project_id)) \
if project_id else fb.listFixFors()
# Fetching the issues
self.cancellation_point()
new_issues = []
for milestone in milestones.fixfors.childGenerator():
issues = fb.search(
q='milestone:"%s"' % milestone.sfixfor.text,
cols='sTitle,sStatus,dtLastUpdated,fOpen,sProject,sFixFor,'
'sEmailAssignedTo,ixProject,sPriority'
)
for issue in issues.cases.childGenerator():
self.cancellation_point()
if project_id and issue.ixproject.text != str(project_id):
continue
new_issues.append(issue)
self._process_fogbugz_issue(issue)
last_issue_list = self.sync_engine.get_all_remote()
new_issue_list = [str(issue['ixbug']) for issue in new_issues]
for issue_link in set(last_issue_list).difference(set(new_issue_list)):
self.cancellation_point()
# we make sure that the other backends are not modifying the task
# set
with self.datastore.get_backend_mutex():
tid = self.sync_engine.get_local_id(issue_link)
self.datastore.request_task_deletion(tid)
try:
self.sync_engine.break_relationship(remote_id=issue_link)
except KeyError:
pass
return
###############################################################################
### Process tasks #############################################################
###############################################################################
def _process_fogbugz_issue(self, issue):
""" Given a issue object, finds out if it must be synced to a GTG
note and, if so, it carries out the synchronization (by creating or
updating a GTG task, or deleting itself if the related task has been
deleted)
"""
has_task = self.datastore.has_task
action, tid = self.sync_engine.analyze_remote_id(
issue['ixbug'], has_task, lambda b: True
)
Log.debug('Processing fogbugz (%s)' % action)
if action is None:
return
issue = self._prefetch_issue_data(issue)
with self.datastore.get_backend_mutex():
if action == SyncEngine.ADD:
tid = str(uuid.uuid4())
task = self.datastore.task_factory(tid)
self._populate_task(task, issue)
self.sync_engine.record_relationship(
local_id=tid,
remote_id=str(issue['number']),
meme=SyncMeme(
task.get_modified(),
issue['modified'],
self.get_id()
)
)
self.datastore.push_task(task)
elif action == SyncEngine.UPDATE:
task = self.datastore.get_task(tid)
self._populate_task(task, issue)
meme = self.sync_engine.get_meme_from_remote_id(
issue['number']
)
meme.set_local_last_modified(task.get_modified())
meme.set_remote_last_modified(issue['modified'])
self.save_state()
def _prefetch_issue_data(self, issue):
"""
We fetch all the necessary info that we need from the issue to
populate a task beforehand (these will be used in _populate_task).
@param issue: a fogbugz issue
@returns dict: a dictionary containing the relevant issue attributes
"""
issue_dic = {
'title': issue.stitle.text,
'status': issue.sstatus.text,
'number': issue['ixbug'],
#'text': issue['description'],
'modified': datetime.strptime(
issue.dtlastupdated.text, '%Y-%m-%dT%H:%M:%SZ'
),
'project': issue.sproject.text.replace(' ', '_'),
'milestone': issue.sfixfor.text.replace(' ', '_'),
'completed': issue.fopen.text != 'true',
'assigned_to': issue.semailassignedto.text,
'priority': issue.spriority.text,
}
issue_dic['assigned'] = (
issue_dic['assigned_to'] == self._parameters['username']
)
return issue_dic
def _populate_task(self, task, issue):
""" Fills a GTG task with the data from a fogbugz issue.
@param task: a Task
@param issue: a fogbugz issue
"""
# set task status
if issue["completed"]:
task.set_status(Task.STA_DONE)
else:
task.set_status(Task.STA_ACTIVE)
if task.get_title() != issue['title']:
task.set_title("%s: %s" % (issue["number"], issue['title']))
text = self._build_issue_text(issue)
if task.get_excerpt() != text:
task.set_text(text)
new_tags = set([])
if self._parameters["tag-with-project-name"]:
new_tags = {'@' + issue['project']}
new_tags.add('@' + issue['milestone'])
current_tags = set(task.get_tags_name())
# add the new ones
for tag in new_tags.difference(current_tags):
task.add_tag(tag)
task.add_remote_id(self.get_id(), issue['number'])
def _build_issue_text(self, issue_dic):
""" Creates the text that describes a issue. """
text = 'Priority: ' + issue_dic['priority'] + '\n'
text += _("Link to issue: ") + \
self._parameters['service-url'] + '/f/cases/%s' % \
(issue_dic["number"]) + '\n'
#text += '\n' + issue_dic["text"]
return text
def _login(self):
""" Try logging in. """
try:
self.cancellation_point()
fb = FogBugz(self._parameters['service-url'])
fb.logon(
self._parameters['username'], self._parameters['password']
)
except FogBugzAPIError:
self.quit(disable=True)
BackendSignals().backend_failed(
self.get_id(), BackendSignals.ERRNO_AUTHENTICATION
)
fb = None
self._fb = fb
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment