Task to export calculated numbers of the monthly reports back to HubSpot

To summarize the process we:

- Fetch HubSpot companies with tasks.hubspot.refresh_hubspot_data
- We generate monthly usage reports for thos companies with
  task.organization.schedule_monthly_deal_report

now, with this commit we:

- Push the calculated numbers (unique user and teachers) to the
  corresponding fields in HubSpot companies.

lms/services/hubspot/_client.py

@@ -1,5 +1,22 @@
+import csv
+import json
+import os
+from datetime import date
+from enum import StrEnum
+from logging import getLogger
+from tempfile import NamedTemporaryFile
+
 from hubspot import HubSpot
 
+LOG = getLogger(__name__)
+
+
+class ObjectType(StrEnum):
+    """Hubspot codes for different entity types."""
+
+    # From: https://developers.hubspot.com/docs/api/crm/imports
+    COMPANY = "0-2"
+
 
 class HubSpotClient:
     """A nicer client for the Hubspot API."""
@@ -18,6 +35,66 @@ def get_companies(self):
         ]
         yield from self._get_objects(self._api_client.crm.companies, fields)
 
+    def create_company_import(
+        self, contract_year_billables: list[tuple[str, int, int]]
+    ):
+        """Start an import data from a CSV file to Hubspot to update companies.
+
+        :param contract_year_billables: A list of tuples with: (company id in HubSpot, unique teachers, unique users)
+        """
+        with NamedTemporaryFile(mode="w", suffix=".csv") as csv_file:
+            writer = csv.writer(csv_file)
+            for row in contract_year_billables:
+                writer.writerow(row)
+            # Ensure all rows are written to disk before we start to upload
+            csv_file.flush()
+
+            files = [
+                {
+                    "fileName": os.path.basename(csv_file.name),
+                    "fileFormat": "CSV",
+                    "fileImportPage": {
+                        "hasHeader": False,
+                        "columnMappings": [
+                            {
+                                "columnObjectTypeId": ObjectType.COMPANY,
+                                "columnName": "hs_object_id",
+                                "propertyName": "hs_object_id",
+                                "idColumnType": "HUBSPOT_OBJECT_ID",
+                            },
+                            {
+                                "columnObjectTypeId": ObjectType.COMPANY,
+                                "columnName": "billable_teachers_this_contract_year",
+                                "propertyName": "billable_teachers_this_contract_year",
+                                "idColumnType": None,
+                            },
+                            {
+                                "columnObjectTypeId": ObjectType.COMPANY,
+                                "columnName": "billable_users_this_contract_year",
+                                "propertyName": "billable_users_this_contract_year",
+                                "idColumnType": None,
+                            },
+                        ],
+                    },
+                }
+            ]
+
+            import_request = {
+                "name": f"contract_year_import_{date.today().isoformat()}",
+                "files": files,
+                "dateFormat": "YEAR_MONTH_DAY",
+            }
+
+            LOG.debug(
+                "Creating HubSpot company import with %d rows",
+                len(contract_year_billables),
+            )
+            return self._api_client.crm.imports.core_api.create(
+                import_request=json.dumps(import_request),
+                files=[csv_file.name],
+                async_req=False,
+            )
+
     @classmethod
     def _get_objects(cls, accessor, fields: list[str]):
         return accessor.get_all(properties=fields)

lms/services/hubspot/service.py

@@ -4,7 +4,7 @@
 from sqlalchemy import func, select
 from sqlalchemy.exc import MultipleResultsFound
 
-from lms.models.hubspot import HubSpotCompany
+from lms.models import HubSpotCompany, Organization, OrganizationUsageReport
 from lms.services.hubspot._client import HubSpotClient
 from lms.services.upsert import bulk_upsert
 
@@ -27,27 +27,26 @@ def get_company(self, organization_id: str) -> HubSpotCompany | None:
             # More than one company pointing to the same org is a data entry error, ignore them.
             return None
 
-    def get_companies_with_active_deals(self, date_: date):
+    def _companies_with_active_deals_query(self, date_: date):
         # Exclude companies that map to the same Organization.
         # We allow these on the DB to be able to report on the situation to prompt a human to fix it.
         non_duplicated_companies = (
             select(HubSpotCompany.lms_organization_id)
             .group_by(HubSpotCompany.lms_organization_id)
             .having(func.count(HubSpotCompany.lms_organization_id) == 1)
         )
-        return (
-            self._db.query(HubSpotCompany)
-            .where(
-                # Exclude duplicates
-                HubSpotCompany.lms_organization_id.in_(non_duplicated_companies),
-                # Only companies link to an organization
-                HubSpotCompany.organization != None,  # noqa: E711
-                HubSpotCompany.current_deal_services_start <= date_,
-                HubSpotCompany.current_deal_services_end >= date_,
-            )
-            .all()
+        return select(HubSpotCompany).where(
+            # Exclude duplicates
+            HubSpotCompany.lms_organization_id.in_(non_duplicated_companies),
+            # Only companies with a link to an organization
+            HubSpotCompany.organization != None,  # noqa: E711
+            HubSpotCompany.current_deal_services_start <= date_,
+            HubSpotCompany.current_deal_services_end >= date_,
         )
 
+    def get_companies_with_active_deals(self, date_: date):
+        return self._db.scalars(self._companies_with_active_deals_query(date_)).all()
+
     def refresh_companies(self) -> None:
         """Refresh all companies in the DB upserting accordingly."""
         companies = self._client.get_companies()
@@ -84,6 +83,31 @@ def refresh_companies(self) -> None:
             ],
         )
 
+    def export_companies_contract_billables(self, date_: date):
+        """Export the contract billable numbers to HubSpot."""
+        query = (
+            self._companies_with_active_deals_query(date_)
+            .join(
+                Organization,
+                Organization.public_id == HubSpotCompany.lms_organization_id,
+            )
+            .join(OrganizationUsageReport)
+            .distinct(OrganizationUsageReport.organization_id)
+            .order_by(
+                OrganizationUsageReport.organization_id,
+                OrganizationUsageReport.until.desc(),
+            )
+        ).with_only_columns(
+            HubSpotCompany.hubspot_id,
+            OrganizationUsageReport.unique_teachers,
+            OrganizationUsageReport.unique_users,
+        )
+
+        # From the point of view of HubSpot we are creating an import
+        self._client.create_company_import(
+            self._db.execute(query).all(),
+        )
+
     @classmethod
     def factory(cls, _context, request):
         return cls(

lms/tasks/hubspot.py

@@ -1,3 +1,5 @@
+from datetime import date
+
 from lms.services import HubSpotService
 from lms.tasks.celery import app
 
@@ -10,3 +12,12 @@ def refresh_hubspot_data():
             hs = request.find_service(HubSpotService)
 
             hs.refresh_companies()
+
+
+@app.task
+def export_companies_contract_billables():
+    with app.request_context() as request:  # pylint: disable=no-member
+        with request.tm:
+            hs = request.find_service(HubSpotService)
+
+            hs.export_companies_contract_billables(date.today())

tests/unit/lms/services/hubspot/_client_test.py

@@ -1,4 +1,5 @@
-from unittest.mock import create_autospec
+import json
+from unittest.mock import create_autospec, sentinel
 
 import pytest
 from hubspot import HubSpot
@@ -21,10 +22,69 @@ def test_get_companies(self, api_client, svc):
         )
         assert companies == list(api_client.crm.companies.get_all.return_value)
 
+    def test_create_company_export(self, api_client, svc, csv, NamedTemporaryFile):
+        svc.create_company_import([(sentinel.id, sentinel.teachers, sentinel.users)])
+
+        NamedTemporaryFile.assert_called_once_with(mode="w", suffix=".csv")
+        csv.writer.assert_called_once()
+        csv.writer.return_value.writerow.assert_called_once_with(
+            (sentinel.id, sentinel.teachers, sentinel.users)
+        )
+        api_client.crm.imports.core_api.create.assert_called_once_with(
+            import_request=json.dumps(
+                {
+                    "name": "contract_year_import_2024-07-10",
+                    "files": [
+                        {
+                            "fileName": "IMPORT.csv",
+                            "fileFormat": "CSV",
+                            "fileImportPage": {
+                                "hasHeader": False,
+                                "columnMappings": [
+                                    {
+                                        "columnObjectTypeId": "0-2",
+                                        "columnName": "hs_object_id",
+                                        "propertyName": "hs_object_id",
+                                        "idColumnType": "HUBSPOT_OBJECT_ID",
+                                    },
+                                    {
+                                        "columnObjectTypeId": "0-2",
+                                        "columnName": "billable_teachers_this_contract_year",
+                                        "propertyName": "billable_teachers_this_contract_year",
+                                        "idColumnType": None,
+                                    },
+                                    {
+                                        "columnObjectTypeId": "0-2",
+                                        "columnName": "billable_users_this_contract_year",
+                                        "propertyName": "billable_users_this_contract_year",
+                                        "idColumnType": None,
+                                    },
+                                ],
+                            },
+                        }
+                    ],
+                    "dateFormat": "YEAR_MONTH_DAY",
+                }
+            ),
+            files=["IMPORT.csv"],
+            async_req=False,
+        )
+
     @pytest.fixture
     def api_client(self):
         return create_autospec(HubSpot, spec_set=True, instance=True)
 
+    @pytest.fixture
+    def csv(self, patch):
+        return patch("lms.services.hubspot._client.csv")
+
+    @pytest.fixture
+    def NamedTemporaryFile(self, patch):
+        mock = patch("lms.services.hubspot._client.NamedTemporaryFile")
+        mock.return_value.__enter__.return_value.name = "IMPORT.csv"
+
+        return mock
+
     @pytest.fixture
     def svc(self, api_client):
         return HubSpotClient(api_client=api_client)

tests/unit/lms/services/hubspot/service_test.py

@@ -67,6 +67,40 @@ def test_refresh_companies(self, svc, hubspot_api_client, db_session):
         assert company.name == "COMPANY"
         assert company.hubspot_id == 100
 
+    def test_export_companies_contract_billables(
+        self, svc, hubspot_api_client, db_session
+    ):
+        organization = factories.Organization(public_id="TEST_ORG")
+        company = factories.HubSpotCompany(
+            lms_organization_id=organization.public_id,
+            current_deal_services_start=date(2020, 1, 1),
+            current_deal_services_end=date(2025, 1, 1),
+        )
+        # Two reports for the same org, we'd expect to get the numbers from the most recent one
+        factories.OrganizationUsageReport(
+            organization=organization,
+            tag="test",
+            key="test-older-report",
+            until=date(2022, 1, 1),
+            unique_users=10,
+            unique_teachers=10,
+        )
+        factories.OrganizationUsageReport(
+            organization=organization,
+            tag="test",
+            key="test-recent-report",
+            until=date(2023, 1, 1),
+            unique_users=100,
+            unique_teachers=100,
+        )
+        db_session.flush()
+
+        svc.export_companies_contract_billables(date(2024, 1, 1))
+
+        hubspot_api_client.create_company_import.assert_called_once_with(
+            [(company.hubspot_id, 100, 100)]
+        )
+
     @pytest.mark.parametrize(
         "value,expected",
         [

tests/unit/lms/tasks/hubpost_test.py

@@ -1,8 +1,10 @@
 from contextlib import contextmanager
+from datetime import date
 
 import pytest
+from freezegun import freeze_time
 
-from lms.tasks.hubspot import refresh_hubspot_data
+from lms.tasks.hubspot import export_companies_contract_billables, refresh_hubspot_data
 
 
 def test_refresh_hubspot_data(hubspot_service):
@@ -11,6 +13,15 @@ def test_refresh_hubspot_data(hubspot_service):
     hubspot_service.refresh_companies.assert_called_once()
 
 
+@freeze_time("2022-06-21 12:00:00")
+def test_export_companies_contract_billables(hubspot_service):
+    export_companies_contract_billables()
+
+    hubspot_service.export_companies_contract_billables.assert_called_once_with(
+        date(2022, 6, 21)
+    )
+
+
 @pytest.fixture(autouse=True)
 def app(patch, pyramid_request):
     app = patch("lms.tasks.hubspot.app")