How to Create a Cohort using Webapi in Python to OHDSI Broadsea?

General
1

Hi colleagues,

Could you please suggest the way how to connect using a pure Python/API to connect to the OHDSI Atlas and create a cohort using a JSON cohort file?

I used the code below to connect to the installed Broadsea image locally:

import requests
import json

cohort_json = json.loads(GENERATED_ANSWER_2[0])

ATLAS_URL = "http://127.0.0.1:80/WebAPI"
USERNAME = "XYZ"
PASSWORD = "XYZ"

session = requests.Session()

login_url = f"{ATLAS_URL}/user/login/db"
payload = {"login": USERNAME, "password": PASSWORD}
login_resp = session.post(login_url, data=payload)
print("Login:", login_resp.status_code)

print("Cookies now:", session.cookies.get_dict())

create_url = f"{ATLAS_URL}/cohortdefinition"
cohort_payload = {
    "name": "My Cohort via Cookies",
    "description": "Created using session cookies",
    "expressionType": "SIMPLE_EXPRESSION",
    "expression": cohort_json,
}

resp = session.post(create_url, json=cohort_payload)
print("Create cohort:", resp.status_code, resp.text)

It returns 200 for the login, and 401 for the cohort creating:
Login: 200
Cookies now: {}
Create cohort: 401

When I open that in my browser, I see the following:
Link: http://127.0.0.1/WebAPI/info
Response:
{“version”:“2.15.1”,“buildInfo”:{“artifactVersion”:“WebAPI 2.15.1-SNAPSHOT”,“build”:“NA”,“timestamp”:“Sat Aug 16 05:18:31 UTC 2025”,“branch”:“2.15.1”,“commitId”:“ba50f13a”,“atlasRepositoryInfo”:{“milestoneId”:47,“releaseTag”:““},“webapiRepositoryInfo”:{“milestoneId”:49,“releaseTag”:””}},“configuration”:{“security”:{“samlActivated”:false,“enabled”:true,“samlEnabled”:false},“plugins”:{“atlasgisEnabled”:false},“person”:{“viewDatesPermitted”:false},“heracles”:{“smallCellCount”:“5”}}}

Any help is very appreciated. I know, I may use wrappers on top of R packages, but I don’t want to do it intentionally.

2

When you log in, you will get a bearer token in the header (check your headers from the first login request to DB.

On subsequent requests, you will need to set the ‘Bearer’ header to that token. That should let you pass authentication with that identity.

1 Like
3

Thanks a lot, Chris!

Now, it works!

I am attaching the full workable Python code for the future reference if anybody needs it:

cohort_json_str = '''
{
	"cdmVersionRange" : ">=5.0.0",
	"PrimaryCriteria" : {
		"CriteriaList" : [
			{
				"ConditionOccurrence" : {
					"CodesetId" : 1,
					"ConditionTypeExclude" : false
				}
			}
		],
		"ObservationWindow" : {
			"PriorDays" : 0,
			"PostDays" : 0
		},
		"PrimaryCriteriaLimit" : {
			"Type" : "All"
		}
	},
	"ConceptSets" : [
		{
			"id" : 1,
			"name" : "Cough or Suptum Finding",
			"expression" : {
				"items" : [
					{
						"concept" : {
							"CONCEPT_ID" : 254761,
							"CONCEPT_NAME" : "Cough",
							"STANDARD_CONCEPT" : "S",
							"STANDARD_CONCEPT_CAPTION" : "Standard",
							"INVALID_REASON" : "V",
							"INVALID_REASON_CAPTION" : "Valid",
							"CONCEPT_CODE" : "49727002",
							"DOMAIN_ID" : "Condition",
							"VOCABULARY_ID" : "SNOMED",
							"CONCEPT_CLASS_ID" : "Clinical Finding"
						},
						"isExcluded" : false,
						"includeDescendants" : true,
						"includeMapped" : false
					},
					{
						"concept" : {
							"CONCEPT_ID" : 4089228,
							"CONCEPT_NAME" : "Sputum finding",
							"STANDARD_CONCEPT" : "S",
							"STANDARD_CONCEPT_CAPTION" : "Standard",
							"INVALID_REASON" : "V",
							"INVALID_REASON_CAPTION" : "Valid",
							"CONCEPT_CODE" : "248595008",
							"DOMAIN_ID" : "Condition",
							"VOCABULARY_ID" : "SNOMED",
							"CONCEPT_CLASS_ID" : "Clinical Finding"
						},
						"isExcluded" : false,
						"includeDescendants" : true,
						"includeMapped" : false
					}
				]
			}
		}
	],
	"QualifiedLimit" : {
		"Type" : "First"
	},
	"ExpressionLimit" : {
		"Type" : "All"
	},
	"InclusionRules" : [],
	"EndStrategy" : {
		"DateOffset" : {
			"DateField" : "EndDate",
			"Offset" : 14
		}
	},
	"CensoringCriteria" : [],
	"CollapseSettings" : {
		"CollapseType" : "ERA",
		"EraPad" : 0
	},
	"CensorWindow" : {}
}
'''

cohort_json = json.loads(cohort_json_str)

USERNAME = "admin"
PASSWORD = "password for admin or other user"

import requests
import json

# cohort_json = json.loads(GENERATED_ANSWER_2[0])

ATLAS_URL = "http://127.0.0.1:80/WebAPI"

session = requests.Session()

login_url = f"{ATLAS_URL}/user/login/db"
payload = {"login": USERNAME, "password": PASSWORD}
login_resp = session.post(login_url, data=payload)
print("Login:", login_resp.status_code)
print(login_resp)

token = login_resp.headers['Bearer']
headers = {"Authorization": f"Bearer {token}"}

create_url = f"{ATLAS_URL}/cohortdefinition/"
cohort_payload = {
    "name": "Cohort via API",
    "description": "Created using API",
    "expressionType": "SIMPLE_EXPRESSION",
    "expression": cohort_json,
}

resp = session.post(create_url, headers=headers, json=cohort_payload)
print("Create cohort:", resp.status_code, resp.text)

Result:

image
image2134×232 26.8 KB

4

Very good! the rest-like API should be pretty easy to consume once you get that ball rolling.

1 Like