Coverage for backend/core/auxiliary/views/HealthChecks.py: 38%

1from drf_spectacular.utils import extend_schema 

2from rest_framework.response import Response 

3from rest_framework.decorators import api_view, permission_classes, renderer_classes 

4from rest_framework import status, permissions 

5from django.db import connection 

6 

7@extend_schema(exclude=True) 

8@api_view(['GET']) 

9@permission_classes([permissions.AllowAny]) 

10def get_api_status(request): 

11 return Response({ 

12 'status': 'ok', 

13 }) 

14 

15@extend_schema(exclude=True) 

16@api_view(['GET']) 

17@permission_classes([permissions.AllowAny]) 

18def check_db_connectivity(request): 

19 

20 try: 

21 

22 # Check database connection 

23 with connection.cursor() as cursor: 

24 cursor.execute("SELECT 1") 

25 result = cursor.fetchone() 

26 

27 if result[0] == 1: 

28 dbStatus = "connected" 

29 else: 

30 dbStatus = "error" 

31 

32 if dbStatus != "error": 

33 

34 # Get Database uptime 

35 with connection.cursor() as cursor: 

36 cursor.execute("SELECT date_trunc('second', current_timestamp - pg_postmaster_start_time()) as uptime;") 

37 dbUptime = cursor.fetchone() 

38 

39 # Get Database version 

40 with connection.cursor() as cursor: 

41 dbVersion = cursor.connection.server_version 

42 

43 # Return information on database and api 

44 return Response({ 

45 'status': 'ok', 

46 'database': { 

47 'status': dbStatus, 

48 'uptime': dbUptime, 

49 'version': dbVersion, 

50 }, 

51 }) 

52 

53 except Exception as ex: 

54 

55 return Response({ 

56 'status': 'error', 

57 'error_message': str(ex), 

58 }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)