Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy_utils/types/range.py: 57%

1""" 

2SQLAlchemy-Utils provides wide variety of range data types. All range data 

3types return Interval objects of intervals_ package. In order to use range data 

4types you need to install intervals_ with: 

5 

6:: 

7 

8 pip install intervals 

9 

10 

11Intervals package provides good chunk of additional interval operators that for 

12example psycopg2 range objects do not support. 

13 

14 

15 

16Some good reading for practical interval implementations: 

17 

18https://wiki.postgresql.org/images/f/f0/Range-types.pdf 

19 

20 

21Range type initialization 

22------------------------- 

23 

24:: 

25 

26 

27 

28 from sqlalchemy_utils import IntRangeType 

29 

30 

31 class Event(Base): 

32 __tablename__ = 'user' 

33 id = sa.Column(sa.Integer, autoincrement=True) 

34 name = sa.Column(sa.Unicode(255)) 

35 estimated_number_of_persons = sa.Column(IntRangeType) 

36 

37 

38 

39You can also set a step parameter for range type. The values that are not 

40multipliers of given step will be rounded up to nearest step multiplier. 

41 

42 

43:: 

44 

45 

46 from sqlalchemy_utils import IntRangeType 

47 

48 

49 class Event(Base): 

50 __tablename__ = 'user' 

51 id = sa.Column(sa.Integer, autoincrement=True) 

52 name = sa.Column(sa.Unicode(255)) 

53 estimated_number_of_persons = sa.Column(IntRangeType(step=1000)) 

54 

55 

56 event = Event(estimated_number_of_persons=[100, 1200]) 

57 event.estimated_number_of_persons.lower # 0 

58 event.estimated_number_of_persons.upper # 1000 

59 

60 

61Range type operators 

62-------------------- 

63 

64SQLAlchemy-Utils supports many range type operators. These operators follow the 

65`intervals` package interval coercion rules. 

66 

67So for example when we make a query such as: 

68 

69:: 

70 

71 session.query(Car).filter(Car.price_range == 300) 

72 

73 

74It is essentially the same as: 

75 

76:: 

77 

78 session.query(Car).filter(Car.price_range == DecimalInterval([300, 300])) 

79 

80 

81Comparison operators 

82^^^^^^^^^^^^^^^^^^^^ 

83 

84All range types support all comparison operators (>, >=, ==, !=, <=, <). 

85 

86:: 

87 

88 Car.price_range < [12, 300] 

89 

90 Car.price_range == [12, 300] 

91 

92 Car.price_range < 300 

93 

94 Car.price_range > (300, 500) 

95 

96 # Whether or not range is strictly left of another range 

97 Car.price_range << [300, 500] 

98 

99 # Whether or not range is strictly right of another range 

100 Car.price_range >> [300, 500] 

101 

102 

103 

104Membership operators 

105^^^^^^^^^^^^^^^^^^^^ 

106 

107:: 

108 

109 Car.price_range.contains([300, 500]) 

110 

111 Car.price_range.contained_by([300, 500]) 

112 

113 Car.price_range.in_([[300, 500], [800, 900]]) 

114 

115 ~ Car.price_range.in_([[300, 400], [700, 800]]) 

116 

117 

118Length 

119^^^^^^ 

120 

121SQLAlchemy-Utils provides length property for all range types. The 

122implementation of this property varies on different range types. 

123 

124In the following example we find all cars whose price range's length is more 

125than 500. 

126 

127:: 

128 

129 session.query(Car).filter( 

130 Car.price_range.length > 500 

131 ) 

132 

133 

134 

135.. _intervals: https://github.com/kvesteri/intervals 

136""" 

137 

138from collections.abc import Iterable 

139from datetime import timedelta 

140 

141import sqlalchemy as sa 

142from sqlalchemy import types 

143from sqlalchemy.dialects.postgresql import ( 

144 DATERANGE, 

145 INT4RANGE, 

146 INT8RANGE, 

147 NUMRANGE, 

148 TSRANGE, 

149) 

150 

151from ..exceptions import ImproperlyConfigured 

152from .scalar_coercible import ScalarCoercible 

153 

154intervals = None 

155try: 

156 import intervals 

157except ImportError: 

158 pass 

159 

160 

161class RangeComparator(types.TypeEngine.Comparator): 

162 @classmethod 

163 def coerced_func(cls, func): 

164 def operation(self, other, **kwargs): 

165 other = self.coerce_arg(other) 

166 return getattr(types.TypeEngine.Comparator, func)(self, other, **kwargs) 

167 

168 return operation 

169 

170 def coerce_arg(self, other): 

171 coerced_types = ( 

172 self.type.interval_class.type, 

173 tuple, 

174 list, 

175 str, 

176 ) 

177 

178 if isinstance(other, coerced_types): 

179 return self.type.interval_class(other) 

180 return other 

181 

182 def in_(self, other): 

183 """Determine whether the given interval is contained by another interval.""" 

184 if isinstance(other, Iterable) and not isinstance(other, str): 

185 other = map(self.coerce_arg, other) 

186 return super().in_(other) 

187 

188 def notin_(self, other): 

189 """Determine whether the given interval is not contained by another interval.""" 

190 if isinstance(other, Iterable) and not isinstance(other, str): 

191 other = map(self.coerce_arg, other) 

192 return super().notin_(other) 

193 

194 def __rshift__(self, other, **kwargs): 

195 """ 

196 Returns whether or not given interval is strictly right of another interval. 

197 

198 .. code-block:: 

199 

200 [a, b] >> [c, d] True, if a > d 

201 """ 

202 other = self.coerce_arg(other) 

203 return self.op('>>')(other) 

204 

205 def __lshift__(self, other, **kwargs): 

206 """ 

207 Returns whether or not given interval is strictly left of another interval. 

208 

209 .. code-block:: 

210 

211 [a, b] << [c, d] True, if b < c 

212 """ 

213 other = self.coerce_arg(other) 

214 return self.op('<<')(other) 

215 

216 def contains(self, other, **kwargs): 

217 """Determine whether the given interval contains another interval.""" 

218 other = self.coerce_arg(other) 

219 return self.op('@>')(other) 

220 

221 def contained_by(self, other, **kwargs): 

222 other = self.coerce_arg(other) 

223 return self.op('<@')(other) 

224 

225 

226class DiscreteRangeComparator(RangeComparator): 

227 @property 

228 def length(self): 

229 return sa.func.upper(self.expr) - self.step - sa.func.lower(self.expr) 

230 

231 

232class IntRangeComparator(DiscreteRangeComparator): 

233 step = 1 

234 

235 

236class DateRangeComparator(DiscreteRangeComparator): 

237 step = timedelta(days=1) 

238 

239 

240class ContinuousRangeComparator(RangeComparator): 

241 @property 

242 def length(self): 

243 return sa.func.upper(self.expr) - sa.func.lower(self.expr) 

244 

245 

246funcs = [ 

247 '__eq__', 

248 '__ne__', 

249 '__lt__', 

250 '__le__', 

251 '__gt__', 

252 '__ge__', 

253] 

254 

255 

256for func in funcs: 

257 setattr(RangeComparator, func, RangeComparator.coerced_func(func)) 

258 

259 

260class RangeType(ScalarCoercible, types.TypeDecorator): 

261 comparator_factory = RangeComparator 

262 

263 def __init__(self, *args, **kwargs): 

264 if intervals is None: 

265 raise ImproperlyConfigured('RangeType needs intervals package installed.') 

266 self.step = kwargs.pop('step', None) 

267 super().__init__(*args, **kwargs) 

268 

269 def load_dialect_impl(self, dialect): 

270 if dialect.name == 'postgresql': 

271 # Use the native range type for postgres. 

272 return dialect.type_descriptor(self.impl) 

273 else: 

274 # Other drivers don't have native types. 

275 return dialect.type_descriptor(sa.String(255)) 

276 

277 def process_bind_param(self, value, dialect): 

278 if value is not None: 

279 return str(value) 

280 return value 

281 

282 def process_result_value(self, value, dialect): 

283 if isinstance(value, str): 

284 factory_func = self.interval_class.from_string 

285 else: 

286 factory_func = self.interval_class 

287 if value is not None: 

288 if self.interval_class.step is not None: 

289 return self.canonicalize_result_value( 

290 factory_func(value, step=self.step) 

291 ) 

292 else: 

293 return factory_func(value, step=self.step) 

294 return value 

295 

296 def canonicalize_result_value(self, value): 

297 return intervals.canonicalize(value, True, True) 

298 

299 def _coerce(self, value): 

300 if value is None: 

301 return None 

302 return self.interval_class(value, step=self.step) 

303 

304 

305class IntRangeType(RangeType): 

306 """ 

307 IntRangeType provides way for saving ranges of integers into database. On 

308 PostgreSQL this type maps to native INT4RANGE type while on other drivers 

309 this maps to simple string column. 

310 

311 Example:: 

312 

313 

314 from sqlalchemy_utils import IntRangeType 

315 

316 

317 class Event(Base): 

318 __tablename__ = 'user' 

319 id = sa.Column(sa.Integer, autoincrement=True) 

320 name = sa.Column(sa.Unicode(255)) 

321 estimated_number_of_persons = sa.Column(IntRangeType) 

322 

323 

324 party = Event(name='party') 

325 

326 # we estimate the party to contain minium of 10 persons and at max 

327 # 100 persons 

328 party.estimated_number_of_persons = [10, 100] 

329 

330 print party.estimated_number_of_persons 

331 # '10-100' 

332 

333 

334 IntRangeType returns the values as IntInterval objects. These objects 

335 support many arithmetic operators:: 

336 

337 

338 meeting = Event(name='meeting') 

339 

340 meeting.estimated_number_of_persons = [20, 40] 

341 

342 total = ( 

343 meeting.estimated_number_of_persons + 

344 party.estimated_number_of_persons 

345 ) 

346 print total 

347 # '30-140' 

348 """ 

349 

350 impl = INT4RANGE 

351 comparator_factory = IntRangeComparator 

352 cache_ok = True 

353 

354 def __init__(self, *args, **kwargs): 

355 super().__init__(*args, **kwargs) 

356 self.interval_class = intervals.IntInterval 

357 

358 

359class Int8RangeType(RangeType): 

360 """ 

361 Int8RangeType provides way for saving ranges of 8-byte integers into 

362 database. On PostgreSQL this type maps to native INT8RANGE type while on 

363 other drivers this maps to simple string column. 

364 

365 Example:: 

366 

367 

368 from sqlalchemy_utils import IntRangeType 

369 

370 

371 class Event(Base): 

372 __tablename__ = 'user' 

373 id = sa.Column(sa.Integer, autoincrement=True) 

374 name = sa.Column(sa.Unicode(255)) 

375 estimated_number_of_persons = sa.Column(Int8RangeType) 

376 

377 

378 party = Event(name='party') 

379 

380 # we estimate the party to contain minium of 10 persons and at max 

381 # 100 persons 

382 party.estimated_number_of_persons = [10, 100] 

383 

384 print party.estimated_number_of_persons 

385 # '10-100' 

386 

387 

388 Int8RangeType returns the values as IntInterval objects. These objects 

389 support many arithmetic operators:: 

390 

391 

392 meeting = Event(name='meeting') 

393 

394 meeting.estimated_number_of_persons = [20, 40] 

395 

396 total = ( 

397 meeting.estimated_number_of_persons + 

398 party.estimated_number_of_persons 

399 ) 

400 print total 

401 # '30-140' 

402 """ 

403 

404 impl = INT8RANGE 

405 comparator_factory = IntRangeComparator 

406 cache_ok = True 

407 

408 def __init__(self, *args, **kwargs): 

409 super().__init__(*args, **kwargs) 

410 self.interval_class = intervals.IntInterval 

411 

412 

413class DateRangeType(RangeType): 

414 """ 

415 DateRangeType provides way for saving ranges of dates into database. On 

416 PostgreSQL this type maps to native DATERANGE type while on other drivers 

417 this maps to simple string column. 

418 

419 Example:: 

420 

421 

422 from sqlalchemy_utils import DateRangeType 

423 

424 

425 class Reservation(Base): 

426 __tablename__ = 'user' 

427 id = sa.Column(sa.Integer, autoincrement=True) 

428 room_id = sa.Column(sa.Integer)) 

429 during = sa.Column(DateRangeType) 

430 """ 

431 

432 impl = DATERANGE 

433 comparator_factory = DateRangeComparator 

434 cache_ok = True 

435 

436 def __init__(self, *args, **kwargs): 

437 super().__init__(*args, **kwargs) 

438 self.interval_class = intervals.DateInterval 

439 

440 

441class NumericRangeType(RangeType): 

442 """ 

443 NumericRangeType provides way for saving ranges of decimals into database. 

444 On PostgreSQL this type maps to native NUMRANGE type while on other drivers 

445 this maps to simple string column. 

446 

447 Example:: 

448 

449 

450 from sqlalchemy_utils import NumericRangeType 

451 

452 

453 class Car(Base): 

454 __tablename__ = 'car' 

455 id = sa.Column(sa.Integer, autoincrement=True) 

456 name = sa.Column(sa.Unicode(255))) 

457 price_range = sa.Column(NumericRangeType) 

458 """ 

459 

460 impl = NUMRANGE 

461 comparator_factory = ContinuousRangeComparator 

462 cache_ok = True 

463 

464 def __init__(self, *args, **kwargs): 

465 super().__init__(*args, **kwargs) 

466 self.interval_class = intervals.DecimalInterval 

467 

468 

469class DateTimeRangeType(RangeType): 

470 impl = TSRANGE 

471 comparator_factory = ContinuousRangeComparator 

472 cache_ok = True 

473 

474 def __init__(self, *args, **kwargs): 

475 super().__init__(*args, **kwargs) 

476 self.interval_class = intervals.DateTimeInterval