Coverage for crateweb/consent/lookup_crs.py: 8%

1""" 

2crate_anon/crateweb/consent/lookup_crs.py 

3 

4=============================================================================== 

5 

6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry. 

7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk). 

8 

9 This file is part of CRATE. 

10 

11 CRATE is free software: you can redistribute it and/or modify 

12 it under the terms of the GNU General Public License as published by 

13 the Free Software Foundation, either version 3 of the License, or 

14 (at your option) any later version. 

15 

16 CRATE is distributed in the hope that it will be useful, 

17 but WITHOUT ANY WARRANTY; without even the implied warranty of 

18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

19 GNU General Public License for more details. 

20 

21 You should have received a copy of the GNU General Public License 

22 along with CRATE. If not, see <https://www.gnu.org/licenses/>. 

23 

24=============================================================================== 

25 

26**Function to look up patient details from CPFT's now-defunct 

27CRS (Care Records System) database.** 

28 

29""" 

30 

31from typing import List 

32 

33from cardinal_pythonlib.dbfunc import dictfetchall, dictfetchone 

34from django.db import connections 

35 

36from crate_anon.crateweb.consent.lookup_common import SignatoryTitles 

37from crate_anon.crateweb.consent.models import PatientLookup 

38from crate_anon.crateweb.consent.utils import make_cpft_email_address 

39 

40 

41# ============================================================================= 

42# CPFT Care Records System (CRS) 

43# ============================================================================= 

44 

45 

46def lookup_cpft_crs( 

47 lookup: PatientLookup, decisions: List[str], secret_decisions: List[str] 

48) -> None: 

49 """ 

50 Looks up patient details from the (defunct) CPFT CRS database. 

51 

52 Args: 

53 lookup: a :class:`crate_anon.crateweb.consent.models.PatientLookup` 

54 decisions: list of human-readable decisions; will be modified 

55 secret_decisions: list of human-readable decisions containing secret 

56 (identifiable) information; will be modified 

57 """ 

58 cursor = connections[lookup.source_db].cursor() 

59 # ------------------------------------------------------------------------- 

60 # CRS 1. Fetch basic details 

61 # ------------------------------------------------------------------------- 

62 # Incoming nhs_number will be a number. However, the database has a VARCHAR 

63 # field (nhs_identifier) that may include spaces. So we compare a 

64 # whitespace-stripped field to our value converted to a VARCHAR: 

65 # WHERE REPLACE(nhs_identifier, ' ', '') = CAST(%s AS VARCHAR) 

66 # ... or the other way round: 

67 # WHERE CAST(nhs_identifier AS BIGINT) = %s 

68 cursor.execute( 

69 """ 

70 SELECT 

71 patient_id, -- M number (PK) 

72 -- nhs_identifier, 

73 title, 

74 forename, 

75 surname, 

76 gender, 

77 -- ethnicity, 

78 -- marital_status, 

79 -- religion, 

80 dttm_of_birth, 

81 dttm_of_death 

82 FROM mpi 

83 WHERE CAST(nhs_identifier AS BIGINT) = %s 

84 """, 

85 [lookup.nhs_number], 

86 ) 

87 rows = dictfetchall(cursor) 

88 if not rows: 

89 decisions.append("NHS number not found in mpi table.") 

90 return 

91 if len(rows) > 1: 

92 decisions.append("Two patients found with that NHS number; aborting.") 

93 return 

94 row = rows[0] 

95 crs_patient_id = row["patient_id"] 

96 lookup.pt_local_id_description = "CPFT M number" 

97 lookup.pt_local_id_number = crs_patient_id 

98 secret_decisions.append(f"CPFT M number: {crs_patient_id}.") 

99 lookup.pt_found = True 

100 lookup.pt_title = row["title"] or "" 

101 lookup.pt_first_name = row["forename"] or "" 

102 lookup.pt_last_name = row["surname"] or "" 

103 lookup.pt_sex = row["gender"] or "" 

104 lookup.pt_dob = row["dttm_of_birth"] 

105 lookup.pt_dod = row["dttm_of_death"] 

106 lookup.pt_dead = bool(lookup.pt_dod) 

107 # Deal with dodgy case 

108 lookup.pt_title = lookup.pt_title.title() 

109 lookup.pt_first_name = lookup.pt_first_name.title() 

110 lookup.pt_last_name = lookup.pt_last_name.title() 

111 # ------------------------------------------------------------------------- 

112 # CRS 2. Address 

113 # ------------------------------------------------------------------------- 

114 cursor.execute( 

115 """ 

116 SELECT 

117 -- document_id, -- PK 

118 address1, 

119 address2, 

120 address3, 

121 address4, 

122 postcode, 

123 email 

124 -- startdate 

125 -- enddate 

126 -- patient_id 

127 

128 FROM Address 

129 WHERE 

130 patient_id = %s 

131 AND enddate IS NULL 

132 """, 

133 [crs_patient_id], 

134 ) 

135 row = dictfetchone(cursor) 

136 if not row: 

137 decisions.append("No address found in Address table.") 

138 else: 

139 lookup.pt_address_1 = row["address1"] or "" 

140 lookup.pt_address_2 = row["address2"] or "" 

141 lookup.pt_address_3 = row["address3"] or "" 

142 lookup.pt_address_4 = row["address4"] or "" 

143 lookup.pt_address_6 = row["postcode"] or "" 

144 lookup.pt_email = row["email"] or "" 

145 # ------------------------------------------------------------------------- 

146 # CRS 3. GP 

147 # ------------------------------------------------------------------------- 

148 cursor.execute( 

149 """ 

150 SELECT 

151 -- sourcesystempk, # PK 

152 -- patient_id, # FK 

153 -- national_gp_id, 

154 gpname, 

155 -- national_practice_id, 

156 practicename, 

157 address1, 

158 address2, 

159 address3, 

160 address4, 

161 address5, 

162 postcode, 

163 telno 

164 -- startdate, 

165 -- enddate, 

166 FROM PracticeGP 

167 WHERE 

168 patient_id = %s 

169 AND enddate IS NULL 

170 """, 

171 [crs_patient_id], 

172 ) 

173 row = dictfetchone(cursor) 

174 if not row: 

175 decisions.append("No GP found in PracticeGP table.") 

176 else: 

177 lookup.gp_found = True 

178 lookup.set_gp_name_components( 

179 row["gpname"] or "", decisions, secret_decisions 

180 ) 

181 lookup.gp_address_1 = row["practicename"] or "" 

182 lookup.gp_address_2 = row["address1"] or "" 

183 lookup.gp_address_3 = row["address2"] or "" 

184 lookup.gp_address_4 = row["address3"] or "" 

185 lookup.gp_address_5 = ", ".join( 

186 [row["address4"] or "", row["address5"] or ""] 

187 ) 

188 lookup.gp_address_6 = row["postcode"] 

189 lookup.gp_telephone = row["telno"] 

190 # ------------------------------------------------------------------------- 

191 # CRS 4. Clinician 

192 # ------------------------------------------------------------------------- 

193 cursor.execute( 

194 """ 

195 SELECT 

196 -- patient_id, # PK 

197 -- trustarea, 

198 consultanttitle, 

199 consultantfirstname, 

200 consultantlastname, 

201 carecoordinatortitle, 

202 carecoordinatorfirstname, 

203 carecoordinatorlastname, 

204 carecoordinatoraddress1, 

205 carecoordinatoraddress2, 

206 carecoordinatoraddress3, 

207 carecoordinatortown, 

208 carecoordinatorcounty, 

209 carecoordinatorpostcode, 

210 carecoordinatoremailaddress, 

211 carecoordinatormobilenumber, 

212 carecoordinatorlandlinenumber 

213 FROM CDLPatient 

214 WHERE 

215 patient_id = %s 

216 """, 

217 [crs_patient_id], 

218 ) 

219 row = dictfetchone(cursor) 

220 if not row: 

221 decisions.append("No clinician info found in CDLPatient table.") 

222 else: 

223 lookup.clinician_address_1 = row["carecoordinatoraddress1"] or "" 

224 lookup.clinician_address_2 = row["carecoordinatoraddress2"] or "" 

225 lookup.clinician_address_3 = row["carecoordinatoraddress3"] or "" 

226 lookup.clinician_address_4 = row["carecoordinatortown"] or "" 

227 lookup.clinician_address_5 = row["carecoordinatorcounty"] or "" 

228 lookup.clinician_address_6 = row["carecoordinatorpostcode"] or "" 

229 lookup.clinician_telephone = " / ".join( 

230 [ 

231 row["carecoordinatorlandlinenumber"] or "", 

232 row["carecoordinatormobilenumber"] or "", 

233 ] 

234 ) 

235 careco_email = row[ 

236 "carecoordinatoremailaddress" 

237 ] or make_cpft_email_address( 

238 row["carecoordinatorfirstname"], row["carecoordinatorlastname"] 

239 ) 

240 cons_email = make_cpft_email_address( 

241 row["consultantfirstname"], row["consultantlastname"] 

242 ) 

243 if careco_email: 

244 # Use care coordinator information 

245 lookup.clinician_found = True 

246 lookup.clinician_title = row["carecoordinatortitle"] or "" 

247 lookup.clinician_first_name = row["carecoordinatorfirstname"] or "" 

248 lookup.clinician_last_name = row["carecoordinatorlastname"] or "" 

249 lookup.clinician_email = careco_email 

250 lookup.clinician_signatory_title = SignatoryTitles.CARE_COORDINATOR 

251 decisions.append("Clinician found: care coordinator (CDL).") 

252 elif cons_email: 

253 # Use consultant information 

254 lookup.clinician_found = True 

255 lookup.clinician_title = row["consultanttitle"] or "" 

256 lookup.clinician_first_name = row["consultantfirstname"] or "" 

257 lookup.clinician_last_name = row["consultantlastname"] or "" 

258 lookup.clinician_email = cons_email 

259 lookup.clinician_signatory_title = ( 

260 SignatoryTitles.CONS_PSYCHIATRIST 

261 ) 

262 lookup.clinician_is_consultant = True 

263 decisions.append("Clinician found: consultant psychiatrist (CDL).") 

264 else: 

265 # Don't know 

266 decisions.append( 

267 "No/insufficient clinician information found (CDL)." 

268 )