morpheus-bigquery
Import prefix morpheus_bigquery. Layer MorpheusBigQueryLayer. Used by
enrichment-scraper, morpheus-hubspot, and phone-agent.
Source: libs/morpheus-bigquery/morpheus_bigquery/client.py. The smallest library in the
repository — one function.
create_bigquery_client()
def create_bigquery_client(*, credentials_json: str) -> bigquery.Client:
"""Build a BigQuery client from a service account JSON string."""
info = json.loads(credentials_json)
credentials = service_account.Credentials.from_service_account_info(info)
return bigquery.Client(credentials=credentials, project=credentials.project_id)Credentials are always passed by the consumer — the library never reads an environment variable. The project is taken from the service account itself rather than configured separately, so a credential swap moves the client to the right project automatically.
Usage
The import is deferred into the handler body, not done at module scope, so the module stays importable in a test environment without the layer:
def handle(self, event: ScheduledEvent, context: Any) -> None:
secrets = load_secrets()
if not secrets.bigquery_credentials.strip():
logger.warning("BigQuery credentials not configured — skipping …")
return
from morpheus_bigquery import create_bigquery_client # deferred
bq_client = create_bigquery_client(credentials_json=secrets.bigquery_credentials)
rows = [dict(row) for row in bq_client.query(SOME_QUERY).result()]Every warehouse-backed handler follows this shape: check for credentials, warn and return if absent, then build the client.
Credentials
BIGQUERY_CREDENTIALS in the shared secret — a Google service-account JSON document stored
as a string. Each consuming function declares it as an optional field defaulting to "":
class HubspotSecrets(BaseModel):
bigquery_credentials: str = ""An empty value is a supported state, not an error. It disables every BigQuery-backed handler with a warning:
| Function | Handlers disabled by empty credentials |
|---|---|
morpheus-hubspot |
|
enrichment-scraper |
|
phone-agent |
|
Clearing the key is therefore a single switch that turns off all warehouse-driven segmentation. See kill switches.
Query patterns
Two styles appear in the handlers.
- Literal query, results as dicts
result = client.query(BQ_LOST_DEAL_REENGAGEMENT_QUERY).result()
rows = [dict(row) for row in result]+
Used by all four re-engagement segments. The query is a constant in the function’s
utils/constants.py, selecting from a named view with a LIMIT.
- Parameterised query
from google.cloud import bigquery as bq
job_config = bq.QueryJobConfig(query_parameters=[
bq.ScalarQueryParameter("watermark", "TIMESTAMP", watermark_to_dt(watermark)),
bq.ScalarQueryParameter("row_limit", "INT64", limit),
])
result = client.query(BQ_CONTACT_SYNC_QUERY, job_config=job_config).result()+
Used by the two watermark-driven syncs. Parameters, not interpolation — the watermark comes
from the configs table.
Type coercion
BigQuery TIMESTAMP values arrive as datetime objects, but exports and casts can produce
strings or epoch integers. Every handler carries a defensive coercion helper:
def _to_datetime(value: Any) -> datetime | None:
if value is None:
return None
if isinstance(value, datetime):
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
if isinstance(value, (int, float)):
return datetime.fromtimestamp(value / 1000, tz=timezone.utc) # epoch ms
try:
return datetime.fromisoformat(str(value))
except (ValueError, TypeError):
return NoneNaive datetimes are assumed UTC. An unparseable value becomes None, which the calling
suppression rules interpret as "no known last call" — deliberately the permissive
direction, so a parse failure never silently suppresses a contact.
The helper is duplicated across handlers rather than shared. It is a candidate for this library if a fourth copy appears.
Warehouse layout
| Dataset | Contents |
|---|---|
|
Stitch export of the HubSpot portal — |
|
Mirror of application tables plus the derived views Morpheus queries |
Details: BigQuery warehouse and BigQuery views.