Avoiding Duplicate AI Work: SQLAlchemy Persistence and Fingerprint-Based Caching in FastAPI
When building an AI-powered application, sending every request to an AI provider can create unnecessary work. Users may submit the same information more than once, refresh a page, or update a record without changing the data that affects the result.
During the development of my personal Lead Qualification & Scoring Engine, I worked on this problem by combining SQLAlchemy persistence with a fingerprint-based caching strategy.
The goal was simple:
- Avoid creating duplicate records for the same lead.
- Reuse an existing qualification when the relevant input has not changed.
- Re-run qualification when meaningful lead information changes.
- Keep the same Lead ID when updating an existing lead.
In this article, I’ll explain the design and the implementation behind it.
The Problem: Repeated Requests
Imagine a sales representative submits the following lead:
- Name: John Smith
- Company: Example SaaS
- Industry: Software
- Pain point: Manual lead qualification
The application qualifies the lead and saves the result. If the representative submits the same lead again, creating another database record and repeating the AI analysis is unnecessary.
However, there is an important distinction: if the lead’s buying signal or pain point changes, the application may need to run the qualification process again. I needed a way to distinguish between these two situations.
The Approach: Identity and Fingerprint
I used two separate hashes because they answer two different questions.
Identity key: Is this the same lead? The identity key is calculated from the lead’s name and company. It helps the application find an existing record for the same person at the same company.
Fingerprint: Has the qualification input changed? The fingerprint is calculated from fields that meaningfully affect qualification, such as the industry, company size, pain point, buying signal, and recent activity. The application compares this fingerprint with the one stored in the database.
This distinction is important: lead identity determines which record to look up, while the fingerprint determines whether the stored qualification can be reused.
Creating a Stable Fingerprint with Python
I implemented the fingerprint logic using Python’s hashlib and json modules. Here is a simplified version of the approach:
import hashlib
import json
def normalize(value):
if isinstance(value, str):
value = value.strip().lower()
return value or None
return value
def create_fingerprint(lead_data, fields):
payload = {
field: normalize(lead_data.get(field))
for field in fields
}
stable_json = json.dumps(
payload,
sort_keys=True,
ensure_ascii=False
)
return hashlib.sha256(
stable_json.encode("utf-8")
).hexdigest()
This example illustrates the core technique. In my project, the implementation reads the selected fields from the lead object.
There are three important details here:
Normalization: Trimming whitespace and normalizing string casing means values such as " Software " and "software" produce the same normalized input.
Stable serialization: Using sort_keys=True ensures that dictionary key order does not affect the serialized representation.
SHA-256 hashing: The normalized JSON is converted into a SHA-256 hash. Identical normalized input produces the same fingerprint.
A hash is useful here as a compact comparison value. It does not replace the original lead data, which is stored separately in the database.
Choosing the Fields Carefully
Not every field should necessarily invalidate a cached qualification. For example, my implementation includes fields such as:
- Name and company
- Job title and country
- Industry and company size
- Website and LinkedIn URL
- Pain point and buying signal
- Recent activity
The implementation deliberately excludes custom_offer, which is treated as a display override rather than a qualification input.
This choice depends on the application’s business rules. If a field starts influencing the qualification result, it should be included in the fingerprint. Otherwise, the application could incorrectly reuse a result calculated from older information.
Persisting Lead Data with SQLAlchemy
I used SQLAlchemy to store lead information and qualification results in a database. The lead model includes fields for:
- The identity key and fingerprint
- Original lead information
- Qualification score and classification
- Scoring breakdown and AI assessment
- Recommended offer and next action
- Sales action plan
- Creation and update timestamps
The model stores structured information in database columns. Some result fields, such as scoring breakdowns and sales action plans, use JSON-compatible columns.
I also created repository functions to keep database operations separate from the API route logic. For example, the update operation modifies an existing SQLAlchemy model instance:
def update_lead_record(db, record, data):
for key, value in data.items():
setattr(record, key, value)
db.commit()
db.refresh(record)
return record
The important behavior is that this updates the existing record rather than inserting a new one. The existing database ID is therefore preserved.
In the actual project, the repository functions also handle creating records, retrieving a lead by ID, looking up a lead by identity, and listing recent leads.
Applying the Cache in the FastAPI Endpoint
The main qualification endpoint follows a straightforward decision process.
- Calculate the lead’s identity key.
- Calculate the fingerprint of the qualification-relevant fields.
- Find an existing record using the identity key.
- Compare the stored fingerprint with the newly calculated fingerprint.
- Reuse the stored qualification or run the qualification process, depending on the comparison.
The core decision looks like this:
existing = get_lead_by_identity(
db,
identity_key
)
if existing is not None:
if existing.fingerprint == fingerprint:
# Reuse the stored qualification.
return {
"cached": True,
"lead": _lead_dict(existing),
"qualification": _qualification_dict(existing),
}
# The lead exists, but relevant input changed.
result = analyze_lead(lead)
# Update the existing record with the new result.
else:
# New lead: qualify and create a record.
result = analyze_lead(lead)
This is a simplified illustration of the decision flow; the complete endpoint also builds the response and persists the new qualification results.
The actual implementation returns a cached flag and a qualification_source value. This makes it easier to distinguish between a result reused from the database and one generated through a new qualification run.
What Happens in Each Scenario?
- New person and company: Run qualification and create a record
- Same identity and same fingerprint: Reuse the stored qualification
- Same identity but changed fingerprint: Re-run qualification and update the existing record
- Display-only offer override changes: Do not invalidate the fingerprint solely for that change
This approach keeps the behavior predictable while avoiding unnecessary repeat work.
Engineering Considerations
Implementing a fingerprint-based cache requires more than generating a hash.
Keep the fingerprint definition aligned with business logic. If qualification starts depending on a new field, update the fingerprint field list as well. Otherwise, stale results may be reused.
Handle normalization deliberately. Normalization should reflect the meaning of the data. Lowercasing text is reasonable for many names and descriptive fields, but it should not be applied indiscriminately to every type of value.
Consider changes to the scoring rules. The fingerprint in this implementation represents the lead input. If scoring rules or model behavior change, the same input may produce a different result. For a production system, consider including a scoring-rule version or qualification version in the cache key, or invalidating cached results when those dependencies change.
Think about concurrent requests. Two requests for the same new lead can arrive at nearly the same time. A production implementation should consider database uniqueness constraints, transaction handling, and concurrency so that simultaneous requests do not create duplicate records.
Measure the benefit. Caching can reduce repeated AI calls, but its actual impact depends on how frequently duplicate requests occur. Logging cache hits, cache misses, and qualification duration can help evaluate whether the approach is effective.
What I Learned
This implementation helped me understand that caching is not simply storing a response and returning it later. The difficult part is deciding when a stored result is still valid.
Separating lead identity from qualification fingerprints made the behavior easier to reason about. SQLAlchemy persistence gave the application a consistent place to store both the input data and its qualification results, while FastAPI provided the endpoint through which the workflow runs.
The same ideas can be useful in other applications that process repeated inputs, including document analysis, product recommendations, reporting tools, and AI-assisted workflows.
Conclusion
Building the Lead Qualification & Scoring Engine gave me an opportunity to apply hashing, database persistence, and caching to a practical backend problem. The main lesson is that a useful cache needs a clear definition of identity and a reliable way to determine whether the underlying input has changed.
By combining a stable fingerprint with SQLAlchemy persistence, the application can reuse unchanged qualification results and update existing lead records when relevant information changes.
I’m continuing to improve the project and explore ways to make AI-powered applications more reliable, explainable, and efficient.
This article describes the implementation in my personal Lead Qualification & Scoring Engine
Comments
No comments yet. Start the discussion.