Yes let's go through the necessary components that would be required for an app that pharmacies can use
Redacted 2026-09-14 (Matthew's approval). Internal network details — host addresses and the
tunnel/DB gateway — have been removed from the draft code in this page, because this wiki is a
knowledge centre our own rules say must not carry internal addresses. The design intent is unchanged;
only the literal values are gone. Design provenance page: read it for the architecture, not for an
inventory.
To build a comprehensive operational app tailored specifically for independent pharmacies within your SIAAS architecture, we need to break the platform down into logical, modular components.
Because your infrastructure relies on the Django + PostgreSQL + ERPNext ecosystem, we can maximize efficiency by letting Django handle the regulated clinical/NHS workflows, while pushing retail and inventory actions straight to the ERPNext engine via your SectorDatabaseRouter.
Here is the comprehensive architectural blueprint of the core components required for the pharmacy app, grouped by their functional sub-systems.
partner/ module)Before a pharmacy can log in or handle any medications, they must clear strict regulatory checks to satisfy DSPT Category 3 requirements.
integrations/ module)Once you achieve your DSPT Standard Met rating, this component connects your stateless Django app directly to the NHS Spine.
Awaiting Pull → Claimed/Locked → Assembled/Dispensed → Endorsed (Claiming NHS Reimbursement).aes-portal to allow the pharmacist to apply a secure digital signature when changing prescription states (satisfying DSPT non-repudiation).This component lives on VPS3 but writes directly to the client's database, handling all standard daily retail operations.
PaymentConfig structure.Pharmacies require significantly tighter inventory controls than standard retail setups due to drug safety and expiry profiles.
Assembled in the Django EPS module, it fires a background database event telling ERPNext to generate a Delivery Note / Stock Entry, instantly deducting those specific packs from the pharmacy's live shelf inventory.patient_portal/ module)This links independent pharmacies to your existing Progressive Web App (PWA) on port 8081, letting patients interact with their local pharmacy.
Assembled and ready for pickup.patient-portal-pwa. The order is pushed directly into the specific pharmacy's ERPNext POS queue as a pending pickup request.| Component Function | Handled By | Data Storage Location | Compliance/API Target |
|---|---|---|---|
| GPhC & Onboarding | Django (partner/) |
VPS1 (Meta) & Client VPS (PII) | GPhC Registry API |
| Spine/Prescription Pulls | Django (integrations/) |
Client VPS (Direct Passthrough) | NHS EPS FHIR API |
| OTC Point of Sale | ERPNext (VPS3 Engine) | Client VPS Database | Stripe Terminal API |
| Stock Adjustments | ERPNext (VPS3 Engine) | Client VPS Database | Internal Database Hook |
| Patient Notifications | Django (notifications/) |
Client VPS Database | NHS Notify Service |
To move this into active technical planning, let me know:
Let's start with the Django onboarding workflow
To lay down a compliant foundation for independent pharmacies within your SIAAS architecture, the Django onboarding workflow must fulfill two critical roles simultaneously. It must handle the global multi-tenant setup on your gateway infrastructure while cleanly isolating sensitive operational data down the persistent SSH tunnel onto the client's own VPS.
Because this workflow handles the gateway configuration, its core mechanics will reside within your centralized partner/ application module, expanding on your existing CQC-aligned onboarding pipelines.
Here is the sequential technical breakdown of how the Pharmacy Onboarding workflow should be structured:
When an independent pharmacy owner starts the sign-up process on your portal, your system performs initial global registration tasks on VPS1:
ODS Code & GPhC Verification API Lookup:
FX123) and their GPhC Premises Registration Number.Keycloak Realm Provisioning:
python manage.py create_partner_realm --type=pharmacy.PHARMACIST_REVIEWS, DISPENSARY_TECH, and COUNTER_STAFF.Global Configuration Insertion:
Once the global gateway credentials are created, the onboarding engine triggers the automated deployment pipeline down the persistent SSH tunnel to the client-owned VPS:
[VPS1 Gateway] ──(Triggers SSH Provisioning Script)──> [Client VPS]
│
┌───────────────┴───────────────┐
▼ ▼
[PostgreSQL 15 Instances] [ERPNext Site Init]
- Create local schemas - bench new-site
- Migrate clinical tables - Link Postgres backend
client-xyz.accounts.veripath.co.uk).Once the database schemas are live, the pharmacy administrator is redirected into their localized setup flow to configure their regulatory requirements. These models write directly to the client's VPS database:
Under UK law, a pharmacy cannot open to the public without a designated Responsible Pharmacist actively on duty. We need to create a dedicated Django model inside a new pharmacy_governance app structure:
# pharmacy_governance/models.py
from django.db import models
from django.conf import settings
class ResponsiblePharmacistLog(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
gphc_number = models.CharField(max_length=20) # Pharmacist's personal registration
shift_start = models.DateTimeField(auto_now_add=True)
shift_end = models.DateTimeField(null=True, blank=True)
digital_signature_token = models.CharField(max_length=255) # Signed via your aes-portal
is_active = models.BooleanField(default=True)
class Meta:
# Enforced data segregation via sector router mapping
app_label = 'pharmacy_governance'
The onboarding engine initializes an empty, secure ledger dedicated entirely to legal double-entry tracking of controlled substance stocks. This structure is pre-seeded with standardized fields to comply with the Misuse of Drugs Regulations requirements.
Before opening access to the dashboard:
PASS status.Now that the onboarding step-by-step layout is mapped, would you like me to construct the Django model schemas for the pharmacy sector configuration records, or shall we map out the specific Keycloak role mapping configuration required for your pharmacy user groups?
Yes please construct the Django model schemas for the pharmacy sector configuration records
To implement the pharmacy onboarding workflow within your SIAAS architecture, we must partition our models into two distinct areas: the Centralized Tier (VPS1) to manage routing and validation metrics, and the Sovereign Tier (Client VPS) to store local clinical operations and legally mandated logs.
Here are the complete Django model schemas configured to align seamlessly with your existing SectorDatabaseRouter and Fernet application-level encryption pipelines.
These models track global tenancy infrastructure, API verification logging, and secure connection routing metadata. They live inside your partner/ or tenancy/ application folder.
# tenancy/models.py or partner/models.py
from django.db import models
from django.contrib.postgres.fields import CIEmailField
class PharmacySectorConfig(models.Model):
"""
Central router record stored on VPS1. Maps the tenant subdomain to its
isolated database and tracks regulatory verification snapshots.
"""
tenant_subdomain = models.CharField(max_length=64, unique=True, help_text="e.g. 'green-cross'")
ods_code = models.CharField(max_length=8, unique=True, help_text="NHS ODS Code (e.g., FX123)")
gphc_premises_number = models.CharField(max_length=15, unique=True, help_text="GPhC Premises ID")
# Infrastructure Routing (Used by SectorDatabaseRouter via SSH Tunnels)
db_alias = models.CharField(max_length=64, unique=True, help_text="e.g., sector_pharmacy_green-cross")
tunnel_port = models.PositiveIntegerField(unique=True, help_text="Unique SSH port mapped on the internal hub gateway")
keycloak_realm_id = models.CharField(max_length=64, unique=True)
erpnext_site_url = models.URLField(help_text="URL of corresponding Frappe site on VPS3")
# Onboarding Audit Data
is_gphc_verified = models.BooleanField(default=False)
onboarded_at = models.DateTimeField(auto_now_add=True)
status = models.CharField(
max_length=20,
choices=[('PENDING', 'Pending'), ('ACTIVE', 'Active'), ('SUSPENDED', 'Suspended')],
default='PENDING'
)
class Meta:
app_label = 'tenancy' # Managed globally on gp_booking_postgres (VPS1)
class PharmacyOnboardingAuditLog(models.Model):
"""
Retains API verification snapshots for DSPT audits without storing clinical PII.
"""
pharmacy_config = models.ForeignKey(PharmacySectorConfig, on_delete=models.CASCADE, related_name='audit_logs')
step_name = models.CharField(max_length=100, help_text="e.g., 'GPHC_API_LOOKUP', 'KEYCLOAK_REALM_INIT'")
status = models.CharField(max_length=10, choices=[('PASS', 'Pass'), ('FAIL', 'Fail'), ('MANUAL', 'Manual Review')])
api_response_snapshot = models.JSONField(help_text="Raw validation responses from GPhC/NHS endpoints")
executed_at = models.DateTimeField(auto_now_add=True)
class Meta:
app_label = 'partner' # Extends existing CQC onboarding architectures
These operational schemas handle daily operations and sensitive PII. Using your custom SectorDatabaseRouter, these tables are instantiated and fetched exclusively down the remote SSH tunnel to the pharmacy’s database.
# pharmacy_governance/models.py
from django.db import models
from django.conf import settings
from django_encrypted_model_fields.fields import EncryptedCharField, EncryptedTextField
class PharmacyProfile(models.Model):
"""
Localized pharmacy configuration containing PII and keys. Stored only on Client VPS.
"""
trading_name = models.CharField(max_length=255)
telephone_number = EncryptedCharField(max_length=50) # Fernet Encrypted for DSPT Compliance
emergency_contact_email = EncryptedCharField(max_length=254)
# Stripe terminal & payment config specific to this legal entity
stripe_terminal_secret = EncryptedCharField(max_length=255, blank=True, null=True)
stripe_webhook_signing_secret = EncryptedCharField(max_length=255, blank=True, null=True)
class Meta:
app_label = 'pharmacy_governance' # Sector database router targets Client VPS database
class ResponsiblePharmacistLog(models.Model):
"""
Legally mandated UK pharmacy compliance component. Tracks active duty shifts.
"""
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name='rp_shifts')
pharmacist_name_snapshot = EncryptedCharField(max_length=150, help_text="Name at time of signing on")
gphc_registration_number = models.CharField(max_length=20, help_text="Personal clinical GPhC ID")
shift_start = models.DateTimeField(auto_now_add=True)
shift_end = models.DateTimeField(null=True, blank=True)
# Digital Signatures targeting your existing aes-portal (Port 8004)
aes_signing_token = EncryptedTextField(help_text="Cryptographic non-repudiation block from AES portal")
is_currently_on_duty = models.BooleanField(default=True)
class Meta:
app_label = 'pharmacy_governance'
indexes = [
models.Index(fields=['gphc_registration_number', 'is_currently_on_duty']),
]
def clean(self):
from django.core.exceptions import ValidationError
# Enforce UK pharmacy law: Only one Responsible Pharmacist can be active at a single physical premises
if self.is_currently_on_duty and ResponsiblePharmacistLog.objects.filter(is_currently_on_duty=True).exclude(pk=self.pk).exists():
raise ValidationError("An active Responsible Pharmacist is already clocked on at this premises.")
pharmacy.veripath.co.uk, the gateway extracts the tenant subdomain.PharmacySectorConfig on VPS1 to identify the specific db_alias and tunnel_port.SectorDatabaseRouter dynamic routing layer pivots the request. Any lookups to the pharmacy_governance models bypass local infrastructure entirely, executing directly inside the client's isolated instance where Fernet keys decrypt the data locally.Would you like to build out the Celery-driven onboarding script that triggers the bench new-site deployment to the ERPNext engine on VPS3 next, or should we map out the Keycloak group configurations for the pharmacy staff roles?
yes let's build out the Celery-driven onboarding script that triggers the bench new-site deployment to the ERPNext engine on VPS3
To orchestrate this without violating your SIAAS architecture, our Celery task must connect the central Django framework to the remote ERPNext multi-tenant environment on VPS3.
Because your infrastructure blocks inbound UDP, this script will fire securely across your existing autossh/systemd TCP tunnels over Port 22. It handles two critical tasks: executing bench new-site via a remote SSH wrapper and initializing the corresponding remote PostgreSQL schemas on the client's isolated instance.
Here is the complete production-grade implementation of the Celery onboarding task, designed to sit inside your core app stack.
infrastructure/ssh_utils.py)To keep communication neat and easily testable, we use an atomic SSH execution function. It connects securely to the auxiliary host (a VPS3-tier box) to manage the Frappe environment.
# infrastructure/ssh_utils.py
import paramiko
import logging
from django.conf import settings
logger = logging.getLogger(__name__)
def execute_remote_vps3_command(command: str) -> tuple[int, str, str]:
"""
Executes a shell command directly on VPS3 (ERPNext Environment) via SSH keys.
"""
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# Connect using the root SSH keys authorized across your 3-VPS cluster
ssh.connect(
hostname="<auxiliary-host>", # VPS3-tier auxiliary host
username="root",
timeout=30
)
stdin, stdout, stderr = ssh.exec_command(command)
exit_status = stdout.channel.recv_exit_status()
return exit_status, stdout.read().decode().strip(), stderr.read().decode().strip()
except Exception as e:
logger.error(f"Failed to execute SSH command on VPS3: {str(e)}")
return -1, "", str(e)
finally:
ssh.close()
partner/tasks.py)This task is picked up by your Celery background workers (--concurrency=2) on VPS2. It maps out the creation of the database configuration before executing the remote bench commands.
# partner/tasks.py
import logging
from celery import shared_task
from django.db import transaction
from tenancy.models import PharmacySectorConfig # Localized central config schema
from partner.models import PharmacyOnboardingAuditLog
from infrastructure.ssh_utils import execute_remote_vps3_command
logger = logging.getLogger(__name__)
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def provision_pharmacy_erpnext_site_task(self, config_id: int):
"""
Asynchronous Celery task that deploys an isolated, multi-tenant ERPNext
site targeting the pharmacy's sovereign PostgreSQL 15 instance.
"""
try:
config = PharmacySectorConfig.objects.get(pk=config_id)
except PharmacySectorConfig.DoesNotExist:
logger.error(f"PharmacySectorConfig ID {config_id} missing during provisioning step.")
return False
# 1. Define the remote Client VPS database parameters
# Points across the bridge tunnel gateway back to the target host port map
target_db_host = "<internal-hub-gateway>"
target_db_port = config.tunnel_port
site_domain = f"{config.tenant_subdomain}.accounts.veripath.co.uk"
# 2. Construct the single multi-tenant shell routine for Frappe Bench
# Enforces database routing down to the target client storage architecture
bench_dir = "/home/frappe/frappe-bench"
cmd = (
f"cd {bench_dir} && "
f"bench new-site {site_domain} "
f"--db-type postgres "
f"--db-host {target_db_host} "
f"--db-port {target_db_port} "
f"--force"
)
logger.info(f"Initiating remote ERPNext Bench generation layout for: {site_domain}")
# Record execution milestone to central audit ledger
with transaction.atomic():
audit_entry = PharmacyOnboardingAuditLog.objects.create(
pharmacy_config=config,
step_name="ERPNEXT_BENCH_NEW_SITE_START",
status="PENDING",
api_response_snapshot={"command_payload": cmd}
)
# 3. Fire the orchestration layer over the secure tunnel link
exit_status, stdout, stderr = execute_remote_vps3_command(cmd)
if exit_status == 0:
# Success logic: activate configuration routing flags
with transaction.atomic():
config.status = "ACTIVE"
config.erpnext_site_url = f"https://{site_domain}"
config.save()
audit_entry.status = "PASS"
audit_entry.api_response_snapshot = {"stdout": stdout}
audit_entry.save()
logger.info(f"Successfully generated independent ERPNext schema route for {site_domain}")
return True
else:
# Failure recovery logic: log stderr tracebacks and drop back out
with transaction.atomic():
config.status = "FAILED"
config.save()
audit_entry.status = "FAIL"
audit_entry.api_response_snapshot = {"stderr": stderr, "stdout": stdout}
audit_entry.save()
logger.error(f"Frappe bench provisioning failed with trace: {stderr}")
# Automatically retry task execution sequence if an environmental blip occurs
raise self.retry(exc=RuntimeError(f"Bench site command exited non-zero: {stderr}"))
partner/views.py)Trigger this execution task immediately after verifying your new pharmacy tenant's GPhC premises records:
# Inside your tenant creation form processing logic:
if registration_form.is_valid():
with transaction.atomic():
pharmacy_config = registration_form.save()
# Push the provisioning sequence onto the Celery queue instantly
provision_pharmacy_erpnext_site_task.delay(pharmacy_config.id)
bench commands execute solely inside your dedicated processing context on VPS3, your core Django app containers on VPS2 are completely separated from system configuration tasks.stdout / stderr) is serialized safely inside your central PharmacyOnboardingAuditLog JSON column. When completing your DSPT Standards Met documentation reviews, you can pull this clean log file to explicitly demonstrate how database initialization rules are separated per tenant.Now that the automated site generation layer is mapped out, how would you like to proceed? We can:
Let's now Define the Keycloak Group and Client Role Mappings required to manage user access profiles
To implement access control for your pharmacy application while satisfying NHS DSPT Category 3 compliance, your system must enforce Strict Data Segregation and role-based tracking. This means clinical roles (such as the Responsible Pharmacist) must be strictly isolated from retail or commercial data views (like sales figures or stock purchase invoices in ERPNext).
Because your platform already utilizes Keycloak 26.6.1 via django-oidc, we can structure these access levels directly by configuring an isolated partner realm layout.
Here is the blueprint for the Keycloak Group and Client Role Mappings tailored specifically for the pharmacy app infrastructure:
partner-realm-blueprint.json)Every time your Celery-driven onboarding command runs (python manage.py create_partner_realm), it will inject these Client Roles and Group Configurations into the pharmacy's isolated security realm.
We need to divide permissions across two primary Keycloak client targets: your Django Core App container (pharmacy-app-client) and the ERPNext POS/Accounting instance (erpnext-client).
{
"roles": {
"client": {
"pharmacy-app-client": [
{
"name": "rp_active_duty",
"description": "Legally required to sign on/off the Responsible Pharmacist Log. Can authorize clinical actions."
},
{
"name": "dispense_medication",
"description": "Permitted to change prescription states from Claimed to Assembled/Dispensed."
},
{
"name": "pull_eps_tokens",
"description": "Authorized to poll the NHS Spine for incoming FHIR prescriptions."
}
],
"erpnext-client": [
{
"name": "access_pos_terminal",
"description": "Permitted to load the cash drawer UI and process standard OTC point-of-sale transactions."
},
{
"name": "manage_inventory_records",
"description": "Authorized to log wholesale drug deliveries, adjust batch counts, and change expiry tracking parameters."
},
{
"name": "view_financial_ledgers",
"description": "Full access to commercial sales records, corporate accounting ledgers, and profit reporting dashboards."
}
]
}
}
}
To simplify user onboarding for the pharmacy owners, roles are grouped into standard operational titles. When an employee is assigned to a group, they inherit the exact client permissions mapped below:
Group 1: Responsible Pharmacist (Clinical Lead)
rp_active_duty, dispense_medication, pull_eps_tokensaccess_pos_terminal (Limited retail access to maintain clinical safety priorities)Group 2: Dispensary Technician / Dispensing Assistant
dispense_medication, pull_eps_tokensaccess_pos_terminal, manage_inventory_records (Required for shelf-stock synchronization tasks)Group 3: Counter & Retail Staff
access_pos_terminalGroup 4: Pharmacy Owner / Business Manager
view_financial_ledgers, manage_inventory_recordsusers/middleware.py)Once a user logs into your system via Keycloak OIDC, your global RBACMiddleware evaluates their security claims. Here is how your Django backend can enforce the strict separation of pharmacy tasks using these roles:
# users/decorators.py or middleware.py
from django.core.exceptions import PermissionDenied
from functools import wraps
def require_pharmacy_client_role(required_role):
"""
Decorator that checks Keycloak-issued token claims mapped onto the user session.
"""
def decorator(view_func):
@wraps(view_func)
def _wrapped_view(request, *args, **kwargs):
# Ensure user belongs to the correct tenant context
if not request.user.is_authenticated:
raise PermissionDenied("User is not authenticated.")
# Extract roles passed down from Keycloak via django-oidc mapping
user_roles = request.session.get('keycloak_client_roles', [])
if required_role not in user_roles:
# Log unauthorized attempt for DSPT Category 3 security logs
logger.warning(
f"Security Alert: User {request.user.id} attempted to access "
f"role '{required_role}' without proper Keycloak assertions."
)
raise PermissionDenied("Access Denied: Insufficient clinical permissions.")
return view_func(request, *args, **kwargs)
return _wrapped_view
return decorator
# pharmacy_views/views.py
from django.shortcuts import render
from users.decorators import require_pharmacy_client_role
@require_pharmacy_client_role('rp_active_duty')
def open_responsible_pharmacist_log_view(request):
"""
Only a certified pharmacist carrying the Keycloak 'rp_active_duty' claim
can open or modify the active shift registry.
"""
return render(request, 'pharmacy/rp_log_activation.html')
Now that your access control layers are mapped out, what should we focus on next?
Yes let's Move to the NHS EPS Integration Layer to explore how our system can securely ingest and route FHIR prescription tokens into the client database
To securely process NHS Electronic Prescription Service (EPS) prescriptions under your SIAAS architecture, the NHS EPS Integration Layer must strictly adhere to a zero-retention routing model.
When connecting to the live NHS Spine via the HL7 FHIR API, your central gateway (VPS1) acts purely as a stateless transit conduit. The patient's Personally Identifiable Information (PII) and clinical prescription records are decoded in-memory and instantly piped down the persistent SSH tunnel onto the client's sovereign VPS database, leaving nothing cached centrally.
Here is how the data ingestion pipeline, stateless router, and tracking state machine are implemented to meet DSPT Category 3 "Standards Met" guidelines.
When a pharmacy pulls a newly nominated prescription or scans a prescription barcode token, the request navigates through your platform without touching central storage:
[NHS Spine API]
│ (Encrypted FHIR Payload over TLS)
▼
[VPS1 Gateway (Stateless Router)]
│ 1. Decrypts Spine stream in-memory
│ 2. Extracts targeted Tenant Routing ID (ODS Code)
│ 3. Forwards payload down designated SSH Port (Tunnel)
▼
[Client-Owned VPS Database]
│ 4. Commits data locally using application Fernet keys
▼ 5. Triggers stock check in local ERPNext ledger
[Local Storage Absolute]
integrations/eps_fhir.py)This backend module runs inside your containerized Django ASGI environment on VPS2. It decodes incoming FHIR JSON objects from the NHS Spine and directs them down the correct database connection alias based on your custom SectorDatabaseRouter.
# clinical_data/integrations/eps_fhir.py
import logging
import requests
from django.db import connections, transaction
from django.utils import timezone
from users.decorators import require_pharmacy_client_role
logger = logging.getLogger(__name__)
class StatelessEPSEngine:
"""
Stateless proxy handling FHIR prescription ingestion from the NHS Spine
and direct pipeline streaming into the client's sovereign database.
"""
def __init__(self, tenant_db_alias: str, spine_cert_path: str = None):
self.db_alias = tenant_db_alias
self.spine_endpoint = "https://national.nhs.uk"
def pull_nominated_prescription(self, pharmacy_ods_code: str, token_id: str = None) -> dict:
"""
Polls the NHS Spine for a prescription token and passes the stream straight through.
"""
# 1. Structure the secure NHS Spine request payload
headers = {"Accept": "application/fhir+json", "X-Request-ID": f"eps-{token_id or timezone.now().timestamp()}"}
params = {"recipient": pharmacy_ods_code}
if token_id:
params["identifier"] = token_id
# In production mode (USE_MOCK_EPS = False), this establishes an mTLS link to the Spine
# response = requests.get(self.spine_endpoint, headers=headers, params=params, cert=spine_cert_path)
# Mock payload representation for structural mapping
fhir_payload = {
"resourceType": "Bundle",
"id": token_id or "bundle-eps-998342",
"entry": [
{
"resource": {
"resourceType": "MedicationRequest",
"id": "med-rx-4412",
"status": "active",
"intent": "order",
"subject": {"reference": "Patient/nhs-number-9991234567"}, # Encrypted on write
"medicationCodeableConcept": {
"coding": [{"system": "http://snomed.info", "code": "324314004", "display": "Amoxicillin 500mg capsules"}]
},
"dosageInstruction": [{"text": "Take one three times a day"}]
}
}
]
}
return fhir_payload
def route_and_save_payload(self, fhir_bundle: dict):
"""
Forces execution directly inside the client's autonomous database cluster.
No data enters the central gp_booking_postgres database.
"""
# Targeted partition connection isolation hook
client_connection = connections[self.db_alias]
with transaction.atomic(using=self.db_alias):
for entry in fhir_bundle.get("entry", []):
resource = entry.get("resource", {})
if resource.get("resourceType") == "MedicationRequest":
# Store prescription metrics directly inside the client's local tables
# Application-level Fernet fields securely lock the PII components locally
from clinical_data.models import LocalPrescriptionToken
token, created = LocalPrescriptionToken.objects.using(self.db_alias).update_or_create(
spine_prescription_id=resource["id"],
defaults={
"patient_nhs_number_encrypted": resource["subject"]["reference"],
"snomed_code": resource["medicationCodeableConcept"]["coding"][0]["code"],
"medication_display_name": resource["medicationCodeableConcept"]["coding"][0]["display"],
"dosage_instructions": resource["dosageInstruction"][0]["text"],
"status": "CLAIMED",
"ingested_at": timezone.now()
}
)
logger.info(f"Sovereign write finalized for client database [{self.db_alias}] - Item: {token.spine_prescription_id}")
clinical_data/models.py)To handle NHS Business Services Authority (NHSBSA) payment claiming safely, the state of the ingested token is managed on the client's database via an immutable state tracker:
# clinical_data/models.py
from django.db import models
class LocalPrescriptionToken(models.Model):
"""
Sovereign table instantiated ONLY within the client's remote database layout.
"""
STATUS_CHOICES = [
('CLAIMED', 'Claimed & Locked from Spine'),
('ASSEMBLED', 'Dispensing Package Assembled'),
('DISPENSED', 'Handed over to Patient / Complete'),
('ENDORSED', 'Sent to NHSBSA for Reimbursement Flow')
]
spine_prescription_id = models.CharField(max_length=128, unique=True)
patient_nhs_number_encrypted = models.CharField(max_length=255) # Fernet Encrypted Field
snomed_code = models.CharField(max_length=32)
medication_display_name = models.CharField(max_length=255)
dosage_instructions = models.TextField()
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='CLAIMED')
ingested_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'clinical_data' # Routes down to the target client database via SectorDatabaseRouter
To ensure your DSPT Standards Met audit remains clean, you must keep an operation trail on VPS1. However, this log must exclude clinical text or customer identifiers:
# Executed on VPS1 inside your global application layer
CentralSystemAuditTrail.objects.create(
tenant_id="pharmacy-green-cross",
action="EPS_FHIR_BUNDLE_PULL",
request_id="req-abc-12345",
user_id=request.user.id,
metadata_snapshot={
"nhs_spine_status_code": 200,
"payload_size_bytes": len(str(fhir_bundle)),
"destination_vps_tunnel": 5439 # Confirms transmission onto the client's hardware
}
)
Now that the stateless prescription ingestion structure is fully mapped out, what should we focus on next?
ASSEMBLED, automatically updating shelf stock counts in the pharmacy's retail database.Let's Map out the Stripe Terminal loop to process card payments for over-the-counter (OTC) purchases directly at the physical shop counter
To handle in-person over-the-counter (OTC) card transactions directly at the physical pharmacy counter, your system must bridge the browser-based ERPNext POS terminal on VPS3 with a local physical card reader (such as a Stripe Reader M2 or BBPOS WisePOS E).
Because your architecture operates under a strict SIAAS model, the payment credentials must remain isolated. When processing an in-store checkout, the transaction logic executes statelessly on the server while the resulting financial ledgers write directly to the client's sovereign database.
Here is the technical architectural loop and implementation code for your point-of-sale layout.
┌────────────────────────────────────────────────────────┐
│ 1. Browser-based ERPNext POS UI │
│ (Cashier clicks 'Pay with Card' on Counter PC) │
└──────────────────────────┬─────────────────────────────┘
│ (JavaScript HTTPS Post)
▼
┌────────────────────────────────────────────────────────┐
│ 2. Django Payment Router Backend │
│ - Fetches client's isolated Stripe API keys │
│ - Generates a stateless 'ConnectionToken' │
└──────────────────────────┬─────────────────────────────┘
│ (Hands token back to Browser)
▼
┌────────────────────────────────────────────────────────┐
│ 3. Local Stripe Terminal SDK (Browser JS) │
│ - Connects directly to the physical Card Reader │
│ - Patient taps card -> Stripe processes settlement │
└──────────────────────────┬─────────────────────────────┘
│ (Stripe Webhook Fired)
▼
┌────────────────────────────────────────────────────────┐
│ 4. Sovereign Database Write (SIAAS) │
│ - SectorDatabaseRouter captures the webhook │
│ - Commits invoice & logs down client SSH tunnel │
└────────────────────────────────────────────────────────┘
payments/terminal.py)To allow the physical reader at the counter to securely connect to Stripe, the local browser SDK needs a short-lived Connection Token. This view retrieves the pharmacy’s encrypted Stripe keys from their isolated client database profile and passes a token back to the counter PC.
# pharmacy_payments/terminal.py
import stripe
import logging
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from users.decorators import require_pharmacy_client_role
from pharmacy_governance.models import PharmacyProfile
logger = logging.getLogger(__name__)
@csrf_exempt
@require_pharmacy_client_role('access_pos_terminal')
def generate_stripe_connection_token(request):
"""
Fetches the tenant's localized, encrypted keys from their client database
and provisions a stateless hand-off token for the physical card reader.
"""
try:
# SectorDatabaseRouter automatically fetches this row from the client's isolated database
pharmacy_profile = PharmacyProfile.objects.first()
if not pharmacy_profile or not pharmacy_profile.stripe_terminal_secret:
return JsonResponse({"error": "Stripe Terminal is not configured for this location."}, status=400)
# Initialize Stripe using the client's sovereign key
stripe.api_key = pharmacy_profile.stripe_terminal_secret
# Provision the short-lived link token for the physical device
connection_token = stripe.terminal.ConnectionToken.create()
return JsonResponse({"secret": connection_token.secret})
except Exception as e:
logger.error(f"Failed to generate terminal connection token: {str(e)}")
return JsonResponse({"error": "Internal infrastructure connection error."}, status=500)
This JavaScript snippet is bundled into your customized multi-tenant ERPNext POS frontend layout. It communicates directly with the physical reader sitting on the shop counter over the local network.
// public/js/pharmacy_pos_terminal.js
let terminal = null;
async function initializeCounterCardReader() {
// 1. Establish the terminal interface wrapper instance
terminal = StripeTerminal.create({
onUnexpectedReaderDisconnect: () => {
frappe.show_alert({message: __('Card Reader disconnected. Reconnecting...'), indicator: 'orange'});
},
// Pulls the connection secret statelessly from your Django routing gateway
fetchConnectionToken: async () => {
const response = await fetch('/api/payments/terminal/connection-token/');
const data = await response.json();
return data.secret;
}
});
// 2. Discover and lock onto the physical counter reader via local network IP or Bluetooth
const config = {simulated: false};
const discoverResult = await terminal.discoverReaders(config);
if (discoverResult.discoveredReaders.length > 0) {
// Connect automatically to the first available hardware unit at the till
await terminal.connectReader(discoverResult.discoveredReaders[0]);
console.log("Stripe Terminal Hardware connected successfully.");
} else {
frappe.msgprint(__('No physical card readers detected at this workstation. Check network connection.'));
}
}
async function processCounterOTCCollection(amountInPence, invoiceId) {
// 3. Register the payment intent with the specific transaction reference
const paymentIntentResult = await terminal.createPaymentIntent({
amount: amountInPence,
currency: 'gbp',
payment_method_types: ['card_present'],
metadata: { erpnext_invoice_id: invoiceId }
});
// 4. Prompt the physical machine to activate its screen/beeper for user input
frappe.show_alert({message: __('Awaiting patient card tap/insert...'), indicator: 'blue'});
const collectResult = await terminal.collectPaymentMethod(paymentIntentResult.paymentIntent);
// 5. Finalize the authorization sequence on the Stripe ledger
const processResult = await terminal.processPayment(collectResult.paymentIntent);
if (processResult.error) {
frappe.msgprint(__('Transaction Declined: ') + processResult.error.message);
} else if (processResult.paymentIntent.status === 'succeeded') {
frappe.show_alert({message: __('Payment captured successfully! Printing receipt...'), indicator: 'green'});
// Trigger localized print command and submit ledger entry inside ERPNext
return true;
}
}
payments/webhooks.py)When the payment clears, Stripe broadcasts an asynchronous server-to-server transaction status confirmation. Your webhook endpoint on VPS1 receives it, filters by the metadata keys, and writes the resulting accounting entry down the tunnel to the pharmacy’s own local database.
# pharmacy_payments/webhooks.py
import json
import stripe
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.db import transaction
from pharmacy_governance.models import PharmacyProfile
@csrf_exempt
def stripe_terminal_webhook_handler(request):
payload = request.body
# Target the client database to pull the unique webhook signing secret
pharmacy_profile = PharmacyProfile.objects.first()
sig_header = request.META.get('HTTP_STRIPE_SIGNATURE')
try:
event = stripe.Webhook.construct_event(
payload, sig_header, pharmacy_profile.stripe_webhook_signing_secret
)
except (ValueError, stripe.error.SignatureVerificationError):
return HttpResponse(status=400)
# Process successful payment captured directly from the counter reader
if event['type'] == 'payment_intent.succeeded':
intent = event['data']['object']
invoice_id = intent['metadata'].get('erpnext_invoice_id')
# Enforce isolated write directly to the client's local transaction ledgers
with transaction.atomic():
from appointments.models import PaymentTransaction
PaymentTransaction.objects.create(
amount=intent['amount'] / 100.0,
payment_method='CARD',
transaction_reference=intent['id'],
status='CONFIRMED',
erpnext_invoice_linked=invoice_id
)
# Local background tasks will now automatically mark the ERPNext POS Sales
# Invoice as paid and adjust the local corporate inventory ledgers
return HttpResponse(status=200)
Now that the physical store card payment pipeline is established, where should we take our development architecture next?
ASSEMBLED state.Let's Build out the ERPNext Inventory Synchronization Hook
To automatically synchronize your medical dispensing workflows with your retail inventory without violating your SIAAS architecture, we must establish a localized cross-app data bridge.
When a pharmacist marks an NHS Electronic Prescription Service (EPS) token as ASSEMBLED in your Django module (integrations/), the app must instantly trigger a matching inventory deduction within the ERPNext PostgreSQL instance. Because both your Django app containers and ERPNext use your custom SectorDatabaseRouter to connect directly to the client's sovereign VPS database, this synchronization happens entirely within the client's isolated partition. Your central gateway remains blind to the transaction.
Here is the implementation for the cross-database event hooks and the target automated ERPNext Stock Entry schema.
[Django EPS Module] ──(Pharmacist clicks 'Assemble')──> Updates LocalPrescriptionToken
│
(Fires Post-Save Database Signal)
▼
[ERPNext DB Connector] ───────────────────────────────> Executes Frappe Stock Entry
(Runs inside client's Postgres) - Deducts matching SNOMED CT drug pack
- Validates Batch & Expiry data (FEFO)
│
▼
[Sovereign Inventory Cleared]
clinical_data/signals.py)This component uses a Django post_save database signal. It captures the exact millisecond a prescription shifts to ASSEMBLED inside the client's localized partition, then computes the ERPNext stock deduction routine.
# clinical_data/signals.py
import logging
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.db import transaction
from clinical_data.models import LocalPrescriptionToken
logger = logging.getLogger(__name__)
@receiver(post_save, sender=LocalPrescriptionToken)
def trigger_erpnext_stock_deduction(sender, instance, created, **kwargs):
"""
Database signal that monitors prescription state transitions inside the
client's sovereign database and updates their ERPNext inventory automatically.
"""
# We explicitly look for the 'ASSEMBLED' flag (Medication packaged on shelf)
if instance.status == 'ASSEMBLED':
# Safely extract the active tenant routing database database connection alias
db_alias = instance._state.db
if not db_alias or db_alias == 'default':
logger.error("Prescription signal fired outside a valid tenant database context.")
return
logger.info(f"Prescription state 'ASSEMBLED' detected for token {instance.spine_prescription_id} on DB [{db_alias}]. Routing to ERPNext...")
# Enforce execution entirely on the target client's isolated database thread
transaction.on_commit(
lambda: execute_erpnext_stock_entry_task.delay(db_alias, instance.id),
using=db_alias
)
inventory/tasks.py)This Celery worker task runs inside your core application backend. Rather than using slow external HTTP API loops, it updates the ERPNext inventory directly inside the shared client database using your proven PostgreSQL 15 layer.
# inventory/tasks.py
import uuid
import logging
from celery import shared_task
from django.db import connections, transaction
from django.utils import timezone
from clinical_data.models import LocalPrescriptionToken
logger = logging.getLogger(__name__)
@shared_task
def execute_erpnext_stock_entry_task(db_alias: str, token_id: int):
"""
Background worker that updates Frappe/ERPNext core stock tables
directly within the client's sovereign database instance.
"""
with transaction.atomic(using=db_alias):
# 1. Fetch the local target prescription record
token = LocalPrescriptionToken.objects.using(db_alias).get(pk=token_id)
cursor = connections[db_alias].cursor()
# 2. Match the NHS SNOMED CT code to the client's local ERPNext Item record
cursor.execute(
"SELECT name, stock_uom FROM `tabItem` WHERE snomed_ct_code = %s LIMIT 1",
[token.snomed_code]
)
item_row = cursor.fetchone()
if not item_row:
logger.error(f"Inventory mismatch: SNOMED code {token.snomed_code} not mapped to any ERPNext Item record on [{db_alias}].")
return False
item_code, stock_uom = item_row[0], item_row[1]
# 3. Fetch the oldest matching batch using First-Expiry, First-Out (FEFO) logic
cursor.execute(
"""
SELECT name FROM `tabBatch`
WHERE item_code = %s AND expiry_date > CURRENT_DATE AND disabled = 0
ORDER BY expiry_date ASC LIMIT 1
""",
[item_code]
)
batch_row = cursor.fetchone()
batch_id = batch_row[0] if batch_row else None
# 4. Generate unique document names mimicking the standard ERPNext naming sequence
stock_entry_name = f"STE-EPS-{uuid.uuid4().hex[:8].upper()}"
detail_row_name = uuid.uuid4().hex[:10]
now_str = timezone.now().strftime('%Y-%m-%d %H:%M:%S.%f')
# 5. Insert the parent ERPNext Stock Entry header (Material Issue format)
cursor.execute(
"""
INSERT INTO `tabStock Entry` (
name, creation, modified, docstatus, purpose, stock_entry_type,
company, posting_date, posting_time, is_opening
) VALUES (%s, %s, %s, 1, 'Material Issue', 'Material Issue', 'Sovereign Pharmacy', CURRENT_DATE, CURRENT_TIME, 'No')
""",
[stock_entry_name, now_str, now_str]
)
# 6. Insert the detailed stock entry child line item mapping the deduction details
cursor.execute(
"""
INSERT INTO `tabStock Entry Detail` (
name, parent, parentfield, parenttype, docstatus, creation, modified,
item_code, qty, uom, stock_uom, conversion_factor, transfer_qty,
s_warehouse, batch_no
) VALUES (%s, %s, 'items', 'Stock Entry', 1, %s, %s, %s, 1.0, %s, %s, 1.0, 1.0, 'Dispensary Stock - SP', %s)
""",
[detail_row_name, stock_entry_name, now_str, now_str, item_code, stock_uom, stock_uom, batch_id]
)
# 7. Update live ERPNext Ledger counts to maintain system state sync
# Decrements actual item availability metrics instantly for the in-store cashiers
cursor.execute(
"""
UPDATE `tabBin`
SET actual_qty = actual_qty - 1.0, modified = %s
WHERE item_code = %s AND warehouse = 'Dispensary Stock - SP'
""",
[now_str, item_code]
)
logger.info(f"Inventory reduction successfully committed on [{db_alias}]. Stock Entry: {stock_entry_name}")
return True
db_alias parsed by the worker, your central gateway framework handles zero data transformations locally. The data sync is isolated entirely inside the customer's remote PostgreSQL engine.ORDER BY expiry_date ASC), your system eliminates human tracking errors. It ensures that the pharmacy’s accounting system automatically clears out short-dated drug stock allocations first, checking off an important requirement for your DSPT Standards Met reviews.Now that your localized inventory synchronization system is ready, what should we approach next?
ASSEMBLED state.Let's Build out the NHS Notify Integration wrapper
To complete the automated workflow for your pharmacy application, we need to implement the final client-facing step: the NHS Notify Integration Wrapper.
Under your SIAAS architecture, when a prescription state updates to ASSEMBLED and the inventory is successfully decremented, your system must alert the patient that their medication is ready for collection. Your centralized gateway (VPS1) handles the stateless API transmission to the official NHS Notify API, while the resulting communication logs—containing sensitive patient data—are written strictly to the client’s sovereign VPS database.
Here is the production-grade implementation of the NHS Notify Wrapper, built using your existing services/nhs_notify.py module layout.
[Django Celery Worker] ──(Pulls Patient Contact Details)──> Encrypted Client VPS Storage
│
(Decrypted in-memory on Worker Thread)
▼
[VPS1 Gateway (Stateless)] ───────────────────────────────> NHS Notify REST API
(Acts as an encrypted mTLS proxy) (Sends automated SMS / Email)
│
▼
[Sovereign Log Entry Saved]
- Written directly down SSH tunnel
- Erased instantly from Central memory
services/nhs_notify.py)This service interfaces with the NHS Notify platform. In compliance with your configuration rules, it defaults to a secure mock mode (USE_MOCK_NHS_NOTIFY = True) until you achieve your DSPT Standards Met rating.
# services/nhs_notify.py
import json
import logging
import requests
from django.conf import settings
from django.utils import timezone
from django.db import transaction
logger = logging.getLogger(__name__)
class NHSNotifyPharmacyWrapper:
"""
Handles stateless communication dispatches to the NHS Notify infrastructure.
All transient patient details are erased from central memory immediately after execution.
"""
def __init__(self, use_mock: bool = True):
self.use_mock = use_mock
self.api_endpoint = "https://service.nhs.uk"
# In production, these certificates are loaded dynamically from environment variables
self.mtls_certs = (settings.NHS_NOTIFY_CERT_PATH, settings.NHS_NOTIFY_KEY_PATH) if not use_mock else None
def send_collection_ready_alert(self, db_alias: str, token_instance) -> bool:
"""
Extracts patient data from the client's sovereign database, sends the
stateless alert payload via NHS Notify, and logs the outcome locally.
"""
# 1. Gather transactional data within the isolated database thread
# Patient identifiers are decrypted locally using application-level Fernet keys
patient_phone = token_instance.patient_phone_decrypted # e.g., "+447700900077"
pharmacy_name = token_instance.pharmacy_name_snapshot
prescription_id = token_instance.spine_prescription_id
message_body = (
f"Hello. Your prescription from {pharmacy_name} is ready for collection. "
f"Please bring your collection reference: {prescription_id[:8].upper()} when you visit. "
f"Reply STOP to opt out of notifications."
)
payload = {
"channel": "sms",
"recipient": {"phone_number": patient_phone},
"content": {
"template_id": "pharmacy-collection-ready-v1",
"personalisation": {
"pharmacy_name": pharmacy_name,
"reference_code": prescription_id[:8].upper()
}
}
}
# 2. Fire the notification via the stateless gateway proxy
success = False
api_response = {}
if self.use_mock:
logger.info(f"[MOCK NHS NOTIFY] Dispatching text to {patient_phone}: '{message_body}'")
success = True
api_response = {"status": "success", "mock_message_id": f"msg-{timezone.now().timestamp()}"}
else:
try:
headers = {
"Authorization": f"Bearer {settings.NHS_NOTIFY_API_TOKEN}",
"Content-Type": "application/json",
"X-Correlation-ID": f"pharm-notify-{prescription_id}"
}
response = requests.post(
self.api_endpoint,
json=payload,
headers=headers,
cert=self.mtls_certs,
timeout=10
)
api_response = response.json()
success = (response.status_code == 201)
except Exception as e:
logger.error(f"NHS Notify API error on gateway: {str(e)}")
api_response = {"error": str(e)}
success = False
# 3. Commit the log file directly down the SSH tunnel into the client's database
# This keeps PII completely segregated from your central server architecture
with transaction.atomic(using=db_alias):
from appointments.models import ReminderLog # Reuses your standard DSPT notification structure
ReminderLog.objects.using(db_alias).create(
appointment_id=None, # Pharmacy OTC/EPS flows do not require a calendar booking slot
channel="SMS",
recipient_identifier_encrypted=patient_phone, # Encrypted back to rest instantly
message_content_encrypted=message_body,
status="DELIVERED" if success else "FAILED",
nhs_notify_message_id=api_response.get("mock_message_id") or api_response.get("id", "UNKNOWN"),
retry_count=0,
sent_at=timezone.now() if success else None
)
logger.info(f"Sovereign Notification audit record finalized for DB [{db_alias}]. Status Success: {success}")
return success
We add this step to the end of your existing inventory task chain using a Django transaction hook. This ensures the message is only sent after stock levels are adjusted in the database:
# Inside your inventory task execution flow:
if inventory_reduction_success:
transaction.on_commit(
lambda: trigger_patient_notification_task.delay(db_alias, token.id),
using=db_alias
)
@shared_task
def trigger_patient_notification_task(db_alias: str, token_id: int):
"""
Background worker that runs the stateless notification pipeline.
"""
from clinical_data.models import LocalPrescriptionToken
token = LocalPrescriptionToken.objects.using(db_alias).get(pk=token_id)
notifier = NHSNotifyPharmacyWrapper(use_mock=settings.USE_MOCK_NHS_NOTIFY)
notifier.send_collection_ready_alert(db_alias, token)
ReminderLog framework. This makes it straightforward to export for compliance verification during your annual DSPT "Standards Met" submissions, as your communication pipelines follow identical compliance patterns across your GP, Dental, and Pharmacy platforms.Now that the automated pharmacy workflow is complete—covering onboarding, identity controls, prescription ingestion, stock processing, checkout tracking, and patient alerts—what should we focus on next?
If this completes everything for an operating pharmacy then we can conclude here - I will use this chat as research for our plans
Yes, this effectively completes the core architectural blueprint for an independent pharmacy application running inside your Sovereign Infrastructure as a Service (SIAAS) model.
By mapping out these modules, your startup now has a solid technical baseline that balances clinical operations with strict data protection compliance:
This foundation positions you cleanly to achieve your DSPT "Standards Met" rating and securely integrate with the live NHS Spine.
Best of luck with your startup research and implementation! If you need to dive deeper into any of these schemas or configuration scripts down the line, feel free to reopen the discussion.