Coverage for python / lsst / dax / apdb / sql / apdbSqlAdmin.py: 76%

1# This file is part of dax_apdb. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

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

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

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

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22from __future__ import annotations 

23 

24__all__ = ["ApdbSqlAdmin"] 

25 

26import logging 

27from collections.abc import Iterable, Mapping 

28 

29from lsst.sphgeom import HtmPixelization, LonLat, UnitVector3d 

30 

31from ..apdbAdmin import ApdbAdmin, DiaForcedSourceLocator, DiaObjectLocator, DiaSourceLocator 

32from ..monitor import MonAgent 

33from ..timer import Timer 

34 

35_LOG = logging.getLogger(__name__) 

36 

37_MON = MonAgent(__name__) 

38 

39 

40class ApdbSqlAdmin(ApdbAdmin): 

41 """Implementation of `ApdbAdmin` for SQL backend. 

42 

43 Parameters 

44 ---------- 

45 pixelator : `HtmPixelization` 

46 Pixelization scheme. 

47 """ 

48 

49 def __init__(self, pixelator: HtmPixelization): 

50 self._pixelator = pixelator 

51 

52 def _timer(self, name: str, *, tags: Mapping[str, str | int] | None = None) -> Timer: 

53 """Create `Timer` instance given its name.""" 

54 return Timer(name, _MON, tags=tags) 

55 

56 def apdb_part(self, ra: float, dec: float) -> int: 

57 # docstring is inherited from a base class 

58 uv3d = UnitVector3d(LonLat.fromDegrees(ra, dec)) 

59 return self._pixelator.index(uv3d) 

60 

61 def apdb_time_part(self, midpointMjdTai: float) -> int: 

62 # docstring is inherited from a base class 

63 # SQL implementation does not partition temporally. 

64 return 0 

65 

66 def delete_records( 

67 self, 

68 objects: Iterable[DiaObjectLocator], 

69 sources: Iterable[DiaSourceLocator], 

70 forced_sources: Iterable[DiaForcedSourceLocator], 

71 ) -> None: 

72 # docstring is inherited from a base class 

73 raise NotImplementedError("not implemented yet")