MongoDB Sync¶

Sync PropertyMe data to MongoDB for analytics, AI/ML pipelines, and system integrations.

Installation¶

uv add pypropertyme[mongodb]

Configuration¶

Set these environment variables (in .env or your shell):

Variable	Description
`MONGODB_URI`	MongoDB connection string (e.g., `mongodb://localhost:27017`)
`MONGODB_DATABASE`	Database name

Quick Start¶

# 1. Set environment variables
export MONGODB_URI="mongodb://localhost:27017"
export MONGODB_DATABASE="pypropertyme"

# 2. Sync data
pypropertyme-mongodb sync

# 3. Check document counts
pypropertyme-mongodb status

CLI Commands¶

pypropertyme-mongodb sync¶

Syncs all records from PropertyMe to MongoDB.

pypropertyme-mongodb sync [--dry-run] [--collection <COLLECTION>]

Option	Description
`--dry-run`	Preview sync without writing to MongoDB
`--collection`	Sync specific collection only (e.g., `contacts`, `properties`)

pypropertyme-mongodb status¶

Shows collection statistics and document counts.

pypropertyme-mongodb status

pypropertyme-mongodb test-connection¶

Tests connections to both PropertyMe and MongoDB.

pypropertyme-mongodb test-connection

Collections¶

The sync creates 8 collections in MongoDB:

Collection	Description
contacts	Owners, tenants, suppliers
members	Agency staff
properties	Managed properties
tenancies	Lease agreements (dates, rent, bond, notes)
tenancy_balances	Financial data (rent, arrears, bond)
jobs	Maintenance jobs
inspections	Property inspections
tasks	Tasks and reminders

Note

tenancies and tenancy_balances are two views of the same PropertyMe entity (same IDs): /tenancies returns active tenancies with lease agreement fields, /tenancies/balances returns all tenancies (active and closed) with added financial fields. Both are mirrored for endpoint fidelity.

How It Works¶

Snapshot-First Fetch¶

Every run starts by fetching all PropertyMe bulk endpoints into one in-memory snapshot, before anything is written to MongoDB. If any fetch fails, the run aborts with the database untouched and the CLI exits non-zero.

Document Structure¶

Documents are stored as complete Pydantic model JSON with no transformations:

{
  "_id": ObjectId("..."),
  "id": "abea0092-b4f4-c250-9bc4-eba5ca44a35b",
  "name_text": "Gary Jarrel",
  "email": "[email protected]",
  "is_owner": true,
  "roles": ["owner"]
}

Upsert Logic¶

Uses MongoDB's bulk_write with ReplaceOne operations keyed by the id field. Documents are created if new, replaced if existing.

Stale Record Refresh¶

After syncing, documents in MongoDB that weren't included in the PropertyMe sync response are refreshed:

Stale properties are replaced from the snapshot's archived properties, and stale tenancies are flagged closed using the snapshot's balances as the close signal
Other collections fetch each stale document's current state from PropertyMe by ID; documents that return 404 (deleted from PropertyMe) are logged and skipped
Documents that have become archived/closed are updated with their true state

Indexing¶

A unique index on the id field is created for each collection to ensure efficient upserts and prevent duplicates.

Module Structure¶

src/pypropertyme/sync/mongodb/
├── __init__.py  # Module exports
├── cli.py       # CLI commands (pypropertyme-mongodb)
└── sync.py      # Core sync logic

Programmatic Usage¶

from pypropertyme.client import Client
from pypropertyme.sync.mongodb import MongoDBSync

# Create PropertyMe client first
pme_client = Client.get_client(token)

# Create sync instance
sync = MongoDBSync(pme_client, "mongodb://localhost:27017", "pypropertyme")

# Run sync
await sync.sync_all()

# Sync specific collection
await sync.sync_collection("contacts")

# Get status
status = await sync.get_status()
print(status)

Querying Data¶

Once synced, you can query the data using MongoDB's query language:

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017")
db = client["pypropertyme"]

# Find all owners
owners = list(db.contacts.find({"is_owner": True}))

# Find properties with rent > 500
high_rent = list(db.properties.find({"rent_amount": {"$gt": 500}}))

# Find open jobs
open_jobs = list(db.jobs.find({"status": {"$in": ["Assigned", "Quoted"]}}))

# Find overdue tenancies
import datetime
overdue = list(db.tenancy_balances.find({"arrears_days": {"$gt": 14}}))

Use Cases¶

Analytics Dashboard¶

# Aggregate rent by property type
pipeline = [
    {"$group": {
        "_id": "$property_type",
        "total_rent": {"$sum": "$rent_amount"},
        "count": {"$sum": 1}
    }}
]
results = list(db.properties.aggregate(pipeline))

AI/ML Integration¶

import pandas as pd

# Export to DataFrame for ML
contacts_df = pd.DataFrame(list(db.contacts.find()))
properties_df = pd.DataFrame(list(db.properties.find()))

# Feature engineering
features = properties_df[["rent_amount", "bedrooms", "bathrooms"]]

Full-Text Search¶

# Create text index
db.contacts.create_index([("name_text", "text"), ("email", "text")])

# Search
results = db.contacts.find({"$text": {"$search": "smith"}})