Python package & cli for syncing between ActionNetwork & AirTable.
To set up a new project with an-at-sync, create a new folder for your project:
mkdir project-name && cd project-nameIn that folder, create a requirements.txt and add an-at-sync as a dependency:
an-at-sync
Install it with pip:
pip install -r requirements.txtCreate a folder for your project namespace:
mkdir project_name
In that folder, create a models.py with this default content:
from datetime import datetime
from typing import Any, Dict, Optional
from an_at_sync.format import convert_adr, standardize_phone
from an_at_sync.model import BaseActivist, BaseEvent, BaseRSVP
from dateutil import tz
from pyairtable.utils import datetime_to_iso_str
from pydantic import HttpUrl, validator
from pydantic.networks import EmailStr
eastern = tz.gettz("America/New_York")
utc = tz.gettz("UTC")
class Activist(BaseActivist):
first_name: Optional[str]
last_name: Optional[str]
email: EmailStr
zip_code: Optional[str]
phone_number: Optional[str]
address: Optional[str]
city: Optional[str]
state: Optional[str]
@classmethod
def from_actionnetwork(cls, source: dict, **kwargs: Any):
address, city, state, zip_code = convert_adr(source["postal_addresses"][0])
return cls(
first_name=source.get("given_name"),
last_name=source.get("family_name"),
email=source["email_addresses"][0]["address"].strip(),
address=address,
city=city,
state=state,
zip_code=zip_code,
phone_number=standardize_phone(source["phone_numbers"][0].get("number")),
)
def display_name(self) -> str:
return f"{self.first_name} {self.last_name}"
def pk(self) -> Dict:
return {"Email": self.email}
def to_airtable(self):
return {
"Email": self.email,
"First Name": self.first_name,
"Last Name": self.last_name,
"Phone": self.phone_number,
"Address": self.address,
"City": self.city,
"State": self.state,
"Zip": self.zip_code,
}
class RSVP(BaseRSVP):
id: str
rsvpd_at: datetime
@classmethod
def from_actionnetwork(cls, source, **kwargs: Any):
return cls(
id=f"{kwargs['activist_record']['id']}-{kwargs['event_record']['id']}",
activist=kwargs["activist"],
event=kwargs["event"],
rsvpd_at=source["created_date"],
)
def display_name(self) -> str:
return f"{self.activist.display_name()} to {self.event.display_name()}"
def pk(self):
return {"Id": self.id}
def to_airtable(self) -> dict:
return {
"Id": self.id,
"RSVP'd At": datetime_to_iso_str(self.rsvpd_at.replace(tzinfo=None)),
}
def activist_column(self) -> str:
return "Volunteer"
def event_column(self) -> str:
return "Event"
class Event(BaseEvent):
url: HttpUrl
name: str
location: str
start_date: datetime
end_date: Optional[datetime]
status: str
@validator("start_date", "end_date")
def dates_must_be_eastern(cls, v: datetime):
return v.replace(tzinfo=eastern) if v is not None else v
@classmethod
def from_actionnetwork(cls, source: dict, **kwargs):
address, city, state, *_ = convert_adr(source["location"])
event_location = "Zoom" if not address else f"{address}, {city}, {state}"
return cls(
url=source["browser_url"],
name=source["title"],
start_date=source["start_date"],
end_date=source.get("end_date"),
status=source["status"],
location=event_location,
)
def display_name(self) -> str:
return self.name
def pk(self) -> Dict:
return {"Url": str(self.url)}
def to_airtable(self) -> dict:
return {
"Url": str(self.url),
"Name": self.name,
"Start At": self.start_date.astimezone(tz=utc)
.replace(tzinfo=None)
.isoformat(timespec="milliseconds")
+ "Z",
"End At": self.end_date.astimezone(tz=utc)
.replace(tzinfo=None)
.isoformat(timespec="milliseconds")
+ "Z"
if self.end_date
else None,
"Status": self.status.capitalize(),
"Location": self.location,
}These models represent the various part of the system we're going to interact independently from the two systems that will use them. The Activist represents an individual person in your campaign. The Event represents a single campaign event, and an RSVP represents a response from an Activist to attend an event.
You can use this as a baseline for your own models, allowing you to customize which fields are synced to AirTable & how.
Next, create this .env file:
AN_AT_SYNC_MODELS="project_name.models"
AN_API_KEY="TODO: ActionNetwork API Key"
AT_API_KEY="TODO: Airtable API Key"
AT_BASE="TODO: Base"
AT_ACTIVISTS_TABLE="TODO: Volunteers Table"
AT_EVENTS_TABLE="TODO: Events Table"
AT_RSVP_TABLE="TODO: RSVP Table"Given the project_name above, the AN_AT_SYNC_MODELS points to the module that an-at-sync can load our custom models from. The rest of the env vars require you to get the proper keys and configure your AirTable account.
Details -> API & Sync. "Your API Key" -> Generate Key. Copy into .env.
Need to create the base first. Create Volunteers, Events, & RSVPs table. Get base id from first part of the URL (starts with app).
User stetings -> Developer Hub -> Personal Access token. data.records:read & data.records:write. Access to the created base. Copy the token, add to .env.
Create fields returned by to_airtable method. Copy from second part of URL and add to .env.
Create fields returned by to_airtable method. Copy from second part of URL and add to .env.
Create fields returned by to_airtable method. Copy from second part of URL and add to .env.
python -m an_at_sync sync events --rsvpsCreate a Dockerfile in your repo:
FROM maadhattah/an-at-sync:latest
COPY ./project_name /app/project_nameThis will copy your configuration onto the Docker image.
Next, deploy it to Fly.io:
fly launchIt will create an app based on the Dockerfile image. Next, import your secrets:
cat .env | fly secrets importLastly, create the webhook in ActionNetwork with the url https://<projet-name>.fly.dev/api/webhooks/actionnetwork.