Coverage for src / augint_library / core.py: 100%

1"""Core library functions for augint-library. 

2 

3This module demonstrates the fundamental pattern for creating library functions 

4that can be used both programmatically and through a CLI. It showcases: 

5- Simple, focused functions with clear purposes 

6- Proper error handling with custom exceptions 

7- Type hints for better IDE support 

8- Comprehensive docstrings with examples 

9 

10Common Use Cases: 

11 1. Direct library usage in Python scripts 

12 2. Building CLIs that wrap library functions 

13 3. Creating testable, reusable components 

14 4. Demonstrating error simulation for testing 

15 

16Example: 

17 Basic library usage: 

18 >>> from augint_library import print_hi, fetch_data 

19 >>> 

20 >>> # Simple function call 

21 >>> print_hi("World") 

22 Hi World 

23 >>> 

24 >>> # Function with error handling 

25 >>> try: 

26 ... data = fetch_data("/api/users", timeout=5.0) 

27 ... print(f"Success: {data['status']}") 

28 ... except NetworkError as e: 

29 ... print(f"Failed: {e.message}") 

30 

31 Building a CLI wrapper: 

32 >>> import click 

33 >>> from augint_library import print_hi 

34 >>> 

35 >>> @click.command() 

36 >>> @click.argument('name') 

37 >>> def greet(name): 

38 ... '''Greet someone.''' 

39 ... print_hi(name) 

40 

41Note: 

42 This module uses simple examples to demonstrate patterns that scale 

43 to complex production libraries. The patterns shown here (error handling, 

44 type hints, documentation) should be applied consistently across all 

45 library modules. 

46""" 

47 

48import random 

49import time 

50from typing import Any 

51 

52from .constants import ( 

53 DEFAULT_API_TIMEOUT, 

54 DEFAULT_FAILURE_RATE, 

55 SIMULATION_DELAY_MIN, 

56 TIMEOUT_FAILURE_MULTIPLIER, 

57 TIMEOUT_SUCCESS_MULTIPLIER, 

58) 

59from .exceptions import NetworkError, NetworkTimeoutError 

60 

61 

62def print_hi(name: str) -> None: 

63 """Print a friendly greeting to the given name. 

64 

65 This is the main library function that can be imported and used 

66 in other Python code. 

67 

68 Args: 

69 name: The name of the person to greet. 

70 

71 Example: 

72 >>> print_hi("Alice") 

73 Hi Alice 

74 """ 

75 print(f"Hi {name}") 

76 

77 

78def fetch_data( 

79 endpoint: str, timeout: float = DEFAULT_API_TIMEOUT, failure_rate: float = DEFAULT_FAILURE_RATE 

80) -> dict[str, Any]: 

81 """Fetch data from a remote endpoint (simulated). 

82 

83 This function simulates an external API call that might fail transiently. 

84 It's designed to demonstrate when retry and circuit breaker patterns are useful. 

85 

86 In a real application, this would make an actual HTTP request to an external service. 

87 The simulation allows us to control failure rates for testing and demonstration. 

88 

89 Args: 

90 endpoint: The API endpoint to fetch from. 

91 timeout: Maximum time to wait for response (seconds). 

92 failure_rate: Probability of failure (0.0-1.0) for simulation. 

93 

94 Returns: 

95 A dictionary containing the fetched data. 

96 

97 Raises: 

98 NetworkTimeoutError: If the request times out. 

99 NetworkError: If the network request fails. 

100 

101 Examples: 

102 Basic usage with guaranteed success: 

103 >>> data = fetch_data("/api/users", timeout=2.0, failure_rate=0.0) 

104 >>> print(data["status"]) 

105 ok 

106 >>> print(data["endpoint"]) 

107 /api/users 

108 

109 Handling timeout errors: 

110 >>> try: 

111 ... data = fetch_data("/api/slow", timeout=0.001) 

112 ... except NetworkTimeoutError as e: 

113 ... print(f"Timeout: {e.message}") 

114 ... print(f"Service: {e.details['service']}") 

115 Timeout: Timeout connecting to API endpoint /api/slow after 0.001s 

116 Service: API endpoint /api/slow 

117 

118 Using with retry decorator for resilience: 

119 >>> from augint_library.resilience import retry 

120 >>> 

121 >>> @retry(max_attempts=3, initial_delay=0.1) 

122 ... def reliable_fetch(endpoint): 

123 ... return fetch_data(endpoint, failure_rate=0.5) 

124 >>> 

125 >>> # This will retry up to 3 times on failure 

126 >>> data = reliable_fetch("/api/data") 

127 

128 Testing error conditions: 

129 >>> # Force a failure for testing 

130 >>> try: 

131 ... data = fetch_data("/api/test", failure_rate=1.0) 

132 ... except NetworkError as e: 

133 ... print(f"Expected failure: {e.code.value}") 

134 Expected failure: NETWORK_ERROR 

135 

136 Note: 

137 This function is intentionally simple to demonstrate patterns. 

138 In production, you would: 

139 - Use actual HTTP libraries (requests, httpx, urllib) 

140 - Add authentication and headers 

141 - Handle different status codes 

142 - Parse response formats (JSON, XML, etc.) 

143 - Add logging and metrics 

144 """ 

145 # Simulate network delay 

146 if failure_rate > 0: 

147 # Allow delays that might exceed timeout when failures are possible 

148 delay = random.uniform(SIMULATION_DELAY_MIN, timeout * TIMEOUT_FAILURE_MULTIPLIER) # noqa: S311 - simulation only 

149 else: 

150 # When failure_rate is 0, ensure we never timeout 

151 delay = random.uniform(SIMULATION_DELAY_MIN, timeout * TIMEOUT_SUCCESS_MULTIPLIER) # noqa: S311 - simulation only 

152 

153 if delay > timeout: 

154 raise NetworkTimeoutError( 

155 service=f"API endpoint {endpoint}", timeout=timeout, attempted_duration=delay 

156 ) 

157 

158 time.sleep(delay) 

159 

160 # Simulate random failures (but not when failure_rate is 0) 

161 if failure_rate > 0 and random.random() < failure_rate: # noqa: S311 - simulation only 

162 raise NetworkError( 

163 f"Failed to connect to {endpoint}", service=f"API endpoint {endpoint}", status_code=503 

164 ) 

165 

166 # Return simulated successful response 

167 return { 

168 "status": "ok", 

169 "endpoint": endpoint, 

170 "data": {"users": ["Alice", "Bob", "Charlie"]}, 

171 "timestamp": time.time(), 

172 }