src/asana_client.py
def sync_tasks(self, linear_project: LinearProject, asana_project_gid: str):
        """Sync Asana tasks with Linear issues."""
        current_app.logger.info(
            f"Syncing tasks for project {asana_project_gid}, linear project {linear_project['slugId']}"
        )

        linear_url_custom_field_gid = current_app.config["ASANA_PROJECTS_CUSTOM_FIELDS"][
            AsanaCustomFieldLabels.LINEAR_URL
        ]

        asana_task_fields = ["gid", "followers", "custom_fields", "completed"]

        current_app.logger.debug(f"fetching project tasks {linear_project['name']}")
        existing_asana_tasks = self.client.tasks.get_tasks_for_project(asana_project_gid, opt_fields=asana_task_fields)
        existing_asana_tasks_by_linear_url = {}
        for existing_asana_task in existing_asana_tasks:
            for custom_field in existing_asana_task["custom_fields"]:
                if custom_field["gid"] == linear_url_custom_field_gid:
                    existing_asana_tasks_by_linear_url[custom_field["text_value"]] = existing_asana_task
        current_app.logger.debug(f"existing tasks: {existing_asana_tasks_by_linear_url}")

        for linear_issue in linear_project["issues"]["nodes"]:
            linear_url_custom_field_text_value = f"https://linear.app/gorgias/issue/{linear_issue['identifier']}"
            current_app.logger.debug(f"sync task for linear issue {linear_issue['identifier']}")

            # linear issue was already synced since there is already an Asana task that has the linear issue attached
            existing_asana_task = existing_asana_tasks_by_linear_url.get(linear_url_custom_field_text_value)
            custom_fields = {linear_url_custom_field_gid: linear_url_custom_field_text_value}
            if not existing_asana_task:
                current_app.logger.debug(f"creating task for linear issue {linear_issue['identifier']}")
                existing_asana_task = self._create_task(asana_project_gid, linear_issue)
                current_app.logger.info(f"Asana task created: {linear_issue['identifier']}")
            else:
                current_app.logger.debug(f"task already exists for linear issue {linear_issue['identifier']}")

            self._update_task(existing_asana_task, linear_issue, custom_fields)
Sync Asana tasks with Linear issues.
Sync Asana tasks with Linear issues.
Sync Asana tasks with Linear issues .
def sync_tasks(self, linear_project: LinearProject, asana_project_gid: str):
        current_app.logger.info(
            f"Syncing tasks for project {asana_project_gid}, linear project {linear_project['slugId']}"
        )
        linear_url_custom_field_gid = current_app.config["ASANA_PROJECTS_CUSTOM_FIELDS"][
            AsanaCustomFieldLabels.LINEAR_URL
        ]
        asana_task_fields = ["gid", "followers", "custom_fields", "completed"]
        current_app.logger.debug(f"fetching project tasks {linear_project['name']}")
        existing_asana_tasks = self.client.tasks.get_tasks_for_project(asana_project_gid, opt_fields=asana_task_fields)
        existing_asana_tasks_by_linear_url = {}
        for existing_asana_task in existing_asana_tasks:
            for custom_field in existing_asana_task["custom_fields"]:
                if custom_field["gid"] == linear_url_custom_field_gid:
                    existing_asana_tasks_by_linear_url[custom_field["text_value"]] = existing_asana_task
        current_app.logger.debug(f"existing tasks: {existing_asana_tasks_by_linear_url}")
        for linear_issue in linear_project["issues"]["nodes"]:
            linear_url_custom_field_text_value = f"https://linear.app/gorgias/issue/{linear_issue['identifier']}"
            current_app.logger.debug(f"sync task for linear issue {linear_issue['identifier']}")
            existing_asana_task = existing_asana_tasks_by_linear_url.get(linear_url_custom_field_text_value)
            custom_fields = {linear_url_custom_field_gid: linear_url_custom_field_text_value}
            if not existing_asana_task:
                current_app.logger.debug(f"creating task for linear issue {linear_issue['identifier']}")
                existing_asana_task = self._create_task(asana_project_gid, linear_issue)
                current_app.logger.info(f"Asana task created: {linear_issue['identifier']}")
            else:
                current_app.logger.debug(f"task already exists for linear issue {linear_issue['identifier']}")
            self._update_task(existing_asana_task, linear_issue, custom_fields)
def sync_tasks ( self , linear_project : LinearProject , asana_project_gid : str ) : current_app . logger . info ( f"Syncing tasks for project {asana_project_gid}, linear project {linear_project['slugId']}" ) linear_url_custom_field_gid = current_app . config [ "ASANA_PROJECTS_CUSTOM_FIELDS" ] [ AsanaCustomFieldLabels . LINEAR_URL ] asana_task_fields = [ "gid" , "followers" , "custom_fields" , "completed" ] current_app . logger . debug ( f"fetching project tasks {linear_project['name']}" ) existing_asana_tasks = self . client . tasks . get_tasks_for_project ( asana_project_gid , opt_fields = asana_task_fields ) existing_asana_tasks_by_linear_url = { } for existing_asana_task in existing_asana_tasks : for custom_field in existing_asana_task [ "custom_fields" ] : if custom_field [ "gid" ] == linear_url_custom_field_gid : existing_asana_tasks_by_linear_url [ custom_field [ "text_value" ] ] = existing_asana_task current_app . logger . debug ( f"existing tasks: {existing_asana_tasks_by_linear_url}" ) for linear_issue in linear_project [ "issues" ] [ "nodes" ] : linear_url_custom_field_text_value = f"https://linear.app/gorgias/issue/{linear_issue['identifier']}" current_app . logger . debug ( f"sync task for linear issue {linear_issue['identifier']}" ) existing_asana_task = existing_asana_tasks_by_linear_url . get ( linear_url_custom_field_text_value ) custom_fields = { linear_url_custom_field_gid : linear_url_custom_field_text_value } if not existing_asana_task : current_app . logger . debug ( f"creating task for linear issue {linear_issue['identifier']}" ) existing_asana_task = self . _create_task ( asana_project_gid , linear_issue ) current_app . logger . info ( f"Asana task created: {linear_issue['identifier']}" ) else : current_app . logger . debug ( f"task already exists for linear issue {linear_issue['identifier']}" ) self . _update_task ( existing_asana_task , linear_issue , custom_fields )
Sync Asana tasks with Linear issues.
Sync Asana tasks with Linear issues .
"""Sync Asana tasks with Linear issues.""" # linear issue was already synced since there is already an Asana task that has the linear issue attached
self
linear_project
LinearProject
asana_project_gid
str