test/shaare/Dw88Zg
test
test
To maintain a clean and organized codebase, it is standard practice to separate your application code from your test code. This is typically achieved by creating a dedicated tests directory at the project's root level.
devops_utils) and the tests directory as siblings.tests directory houses all the test files. It is common for the structure inside tests to mirror the structure of the application package to keep tests organized as the project grows.__init__.py file inside the tests directory to ensure it can be treated as a package if needed, although pytest's discovery mechanism is powerful enough to work without it in most cases.To test your application code, your test files must import the functions, classes, and variables that need to be verified. The standard and most robust way to do this is by using absolute imports that start from the project root.
tests/test_file_ops.py, will use an import statement like from devops_utils.file_ops import check_file_extension.sys.path).ModuleNotFoundError, because the project root is not automatically added to the path in that context. This is why a test runner like pytest is necessary.pytest for Project DiscoveryThe pytest framework is designed to handle the complexities of testing structured projects. When you run pytest from your project's root directory, it intelligently prepares the environment for test execution.
pyproject.toml or pytest.ini), which it uses to identify the project's root.pytest automatically adds this directory to sys.path.python -m pytestWhile running pytest directly is often sufficient, an even more reliable method is to invoke it as a module using the python -m flag. This approach is highly recommended, especially in automated environments like CI/CD pipelines.
python -m pytest guarantees that the current working directory is added to sys.path before pytest begins its execution.When setting up tests for a multi-file project, there are several common issues that can be easily avoided.
ModuleNotFoundError during test runs: This is the most frequent problem and is almost always caused by running pytest from the wrong directory. To avoid this, always run pytest from the project root, or preferably use python -m pytest. An editable install (pip install -e .) provides the most robust solution.conftest.py to define shared fixtures.from ..devops_utils import ... are fragile and should be avoided in test files. Always use absolute imports from the project root (e.g., from devops_utils.file_utils import ...).pyproject.tomlThe Python interpreter doesn't automatically know about our project's structure. The modern and most robust solution is to formally define our project as an installable package. By creating a standard pyproject.toml metadata file, we can perform an "editable install" using the command pip install -e .. This seamlessly links our source code into the virtual environment, making our packages importable from anywhere without manual path hacks or special commands. This is the standard, professional workflow for developing Python applications.
pyproject.toml FileThe pyproject.toml file is the modern, unified standard for configuring Python projects, replacing older files like setup.py.
setuptools, in the [build-system] table.[project] table.pip install -e .An "editable" or "development" install is a special mode for installing packages with pip.
pip install -e . installs the project from the current directory in "editable" mode.site-packages directory back to the original source code..py files are immediately reflected in the installed package without needing to run pip install again.Performing an editable install provides a definitive solution to the path and import issues encountered during development.
sys.path for the entire activated virtual environment.PYTHONPATH environment variable.python -mWhile a script with an if __name__ == "__main__" block can be run with python -m, defining a console script in pyproject.toml is the preferred professional approach.
ping-check) is short and intuitive. A python -m command is long, exposes the internal module path, and is easy to mistype.pyproject.toml file, the command remains the same for the user. The python -m command breaks if you rename or move the target file.pyproject.toml clearly marks it as a public, supported command-line interface for your package.[project]
name = "devops_utils"
version = "0.1.0"
[project.scripts]
check_host = "devops_utils.network_utils.check_host:main"python -m vs. python file.pyThe key to understanding the difference lies in the first entry of sys.path. When Python initializes, it needs to know where to start looking for modules. The way you call it determines this "entry point zero".
When you run a script directly using python path/to/script.py, the interpreter's main task is to execute that specific file. It sets the first entry of sys.path to be the directory that contains the script.
When you run a script as a module using python -m package.module, the interpreter's goal is to locate and run a module within an importable package. It sets the first entry of sys.path to be the current working directory from which the command was executed. This allows absolute imports from the project root to succeed.
While you can run any file with python -m, it can lead to a RuntimeWarning if the file is both a library (meant to be imported) and a script (meant to be run). The best practice is to separate these roles.
file_ops.py and network_ops.py) should contain only reusable functions and classes. They should not contain an if __name__ == "__main__": block for complex script logic.python file.py will often cause ModuleNotFoundError for absolute imports. Avoid this by always running packaged scripts from the project root using python -m.RuntimeWarning. Avoid this by separating concerns: create dedicated runner scripts that import from your library modules.-m. The command must be the full dotted path to the script from the project root (e.g., python -m package.subpackage.script).__init__.py)A Python package provides a way to structure a project's module namespace by using directories. It enables a hierarchical organization of modules, mirroring the file system's structure.
__init__.py is recognized by the Python interpreter as a package..py) but also other subdirectories that are themselves packages.__init__.py__init__.py is the standard and most compatible method.__init__.py file is executed once when the package or any of its modules are first imported, which can be useful for setting up package-level resources. For many packages, this file can simply be left empty.import devops_utils.file_ops. Accessing its contents would then require the full prefix, like devops_utils.file_ops.check_file_extension().from keyword to import the module more directly, as in from devops_utils import file_ops, which then allows access via file_ops.check_file_extension().from devops_utils.file_ops import check_file_extension. This makes the function available to be called directly as check_file_extension().__init__.py to Control Imports__init__.py file itself, you can make them appear as if they belong to the top-level package namespace.To improve organization, we can group related modules into their own dedicated subpackages. A subpackage is a directory inside a parent package that contains its own __init__.py file.
devops_utils, can contain multiple subpackages.file_utils subpackage for file-related modules and a network_utils subpackage for networking modules.__init__.py file (which can be empty) to be recognized by Python as a package.An absolute import provides the complete, explicit path to a module starting from a top-level directory that is on Python's search path (sys.path).
from package.subpackage.module import function.devops_utils package would use from devops_utils.file_utils.file_ops import check_file_extension to access a function.A relative import specifies the path to a module based on the location of the file performing the import. This is done using dot notation.
. refers to the current package, while .. refers to the parent package.from .sibling_module import name brings in a name from a module in the same directory. An import like from ..parent_sibling.module import name navigates one level up and then down into a sibling package.A significant pitfall arises when you attempt to directly execute a Python file that contains relative imports. This action will almost always result in an error.
python devops_utils/network_utils/network_ops.py will raise an ImportError: attempted relative import with no known parent package.__name__) to "__main__" and does not recognize it as being part of a package. Consequently, it cannot resolve relative paths like . or ...import System.py extension can be treated as a module..py suffix. For example, a file named file_ops.py is imported as the file_ops module.import Statementimport module_namemodule_name.function_name.CONFIG in your script will not conflict with module_name.CONFIG.from ... import ....as keyword to rename an imported object, for example, from file_ops import parse_yaml_file as parse_yaml.parse_yaml() instead of file_ops.parse_yaml_file()).my_function and later define your own function with the same name, the original import will be overwritten.from module import * is strongly discouraged because it imports all public names from the module, which can pollute the local namespace and make the code difficult to read and debug.When you execute an import statement, Python needs to locate the corresponding module file. It does this by searching through a specific list of directories.
sys.path, which is part of the standard sys module.sys.path list is automatically populated and typically includes the directory of the script that is currently running, directories specified in the PYTHONPATH environment variable, and the default locations where Python and third-party packages are installed.main.py can seamlessly import utils.py when both files are located in the same folder.Example of main.py
print("Main script starting...")
from devops_utils import (
check_file_extension,
is_host_up,
check_hosts_from_config,
)
import sys
print(sys.path)
filenames = ["config.yaml", "script.sh"]
for filename in filenames:
print(f"Checking {filename}")
print(f"Result: {check_file_extension(filename)}")
print(f"\nIs localhost up? {is_host_up("localhost")}")
print(
f"Is nonexistenthost12345 up? {is_host_up("nonexistenthost12345")}"
)
print(
f"\nAre all hosts from servers_config.yaml up? {check_hosts_from_config("servers_config.yaml")}"
)
Example of file_ops.py
print("Module file_ops is being imported")
from typing import Any
try:
import yaml
except (ModuleNotFoundError, ImportError):
print(
"Warning: PyYAML not found, parse_yaml_file will not work."
)
yaml = None
SUPPORTED_EXTENSIONS: list[str] = [".json", ".yaml", ".txt"]
def check_file_extension(filename: str) -> bool:
"""Checks if a file has a supported extension"""
print(
f" - file_ops.check_file_extension called for {filename}"
)
return any(
filename.endswith(ext) for ext in SUPPORTED_EXTENSIONS
)
def parse_yaml_file(path_str: str) -> dict[str, Any]:
"""Parses a YAML file and returns its contents."""
print(f" - file_ops.parse_yaml_file called for {path_str}")
if yaml:
with open(path_str, "r") as file:
return yaml.safe_load(file)
else:
return {}
unittest.mock module provides tools to create and configure these mock objects and to track interactions.patch function replaces a target object with a mock in a specified scope, either for the duration of a function (decorator) or within a context block (with).patch injects the mock into the test function’s parameters; as a context manager, it yields the mock within the with block.MagicMock instance that you can configure.mock.return_value to define what the mock will return when called.mock.side_effect to simulate an exception being raised by the mock when invoked, to pass different values to be returned by each execution, or to pass a calable to replace the implemented function.assert_called_with and assert_called_once let you verify interactions with the mock.requests.get or requests.post to simulate successful responses, HTTP errors, or timeouts.open() or os.path.exists() to simulate file presence or content.subprocess.run to avoid running real system commands and control return codes.time.sleep or mock datetime.now() to remove delays and make time-based tests deterministic.from unittest.mock import patch, Mock
from pytest_mock import MockerFixture
from dummy_functions import check_file_exists, get_user_data
# Section: Using unittest.mock.patch
def test_check_file_exists_manual_patch() -> None:
filepath = "/path/to/some/file.txt"
patcher = patch("dummy_functions.os.path.exists")
mock_exists = patcher.start()
mock_exists.return_value = True
try:
result = check_file_exists(filepath=filepath)
mock_exists.assert_called_once_with(filepath)
assert result is True
finally:
patcher.stop()
def test_check_file_exists_context_manager() -> None:
filepath = "/path/to/some/file.txt"
with patch("dummy_functions.os.path.exists") as mock_exists:
mock_exists.return_value = True
result = check_file_exists(filepath=filepath)
mock_exists.assert_called_once_with(filepath)
assert result is True
@patch("dummy_functions.os.path.exists")
def test_check_file_exists_decorator(mock_exists: Mock) -> None:
filepath = "/path/to/some/file.txt"
mock_exists.return_value = True
result = check_file_exists(filepath=filepath)
mock_exists.assert_called_once_with(filepath)
assert result is True
def test_check_file_pytest_mocker(mocker: MockerFixture) -> None:
filepath = "/path/to/some/file.txt"
mock_exists = mocker.patch("dummy_functions.os.path.exists")
mock_exists.return_value = True
result = check_file_exists(filepath=filepath)
mock_exists.assert_called_once_with(filepath)
assert result is True
# Section: MagicMock and Configuring Mock Objects
def test_get_user_data_success(mocker: MockerFixture) -> None:
mock_api_response: dict[str, str | int] = {
"id": 1,
"name": "test user",
}
mock_get = mocker.patch("dummy_functions.requests.get")
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = mock_api_response
data = get_user_data(user_id=1)
mock_get.assert_called_once_with(
"https://api.example.com/users/1"
)
assert data == mock_api_response
Example of dummy_function.py
import requests
import os
import subprocess
from typing import Optional, Any
def get_user_data(user_id: str | int) -> dict[str, str | int]:
response = requests.get(
f"https://api.example.com/users/{user_id}"
)
print(f"Status code: {response.status_code}")
response.raise_for_status()
return response.json()
def check_file_exists(filepath: str | os.PathLike[str]) -> bool:
return os.path.exists(filepath)
def get_external_ip():
"""Fetches the current external IP from an external service."""
try:
response = requests.get(
"https://api.ipify.org?format=json", timeout=5
)
response.raise_for_status()
return response.json().get("ip")
except requests.exceptions.RequestException:
return None
def get_current_user() -> Optional[str]:
try:
result = subprocess.run(
["whoami"],
capture_output=True,
text=True,
check=True,
timeout=5,
)
return result.stdout.strip()
except (
subprocess.CalledProcessError,
subprocess.TimeoutExpired,
FileNotFoundError,
):
return None
def fetch_both_endpoints() -> (
tuple[dict[str, Any], dict[str, Any]]
):
"""
Fetch data from two endpoints and return their JSON responses as a tuple.
"""
response2 = requests.get("https://api.example.com/second")
response2.raise_for_status()
data2 = response2.json()
response1 = requests.get("https://api.example.com/first")
response1.raise_for_status()
data1 = response1.json()
return data1, data2
side_effectside_effect attribute on a mock allows you to control its behavior beyond a single return value.side_effect is set to a list, each call to the mock returns the next item in that list, in order.side_effect is a function, it is called with the same arguments as the mock, and its return value is used as the mock’s return.side_effect is an exception, it will raise that exception when the original function is called.Mock: A simple replacement that only creates attributes when accessed, and raises errors for undefined methods or attributes.MagicMock: Inherits from Mock and implements Python’s magic methods (__len__, __enter__, etc.) by default.Mock by default for stubbing external dependencies to catch unintended interactions.MagicMock only when mocking objects that require special behavior, such as context managers or iterables.import subprocess
import pytest
from unittest.mock import MagicMock
from pytest_mock import MockerFixture
from dummy_functions import (
get_current_user,
check_file_exists,
fetch_both_endpoints,
)
# Section: Using side_effect - Exceptions
def test_get_current_user_command_fails(mocker: MockerFixture):
mock_run = mocker.patch("dummy_functions.subprocess.run")
mock_run.side_effect = subprocess.CalledProcessError(
returncode=1, cmd=["whoami"]
)
result = get_current_user()
assert result is None
# Section: Using side_effect - List for Multiple Calls
def test_check_file_exists_side_effect_list(
mocker: MockerFixture,
):
mock_exists = mocker.patch(
"dummy_functions.os.path.exists",
side_effect=[True, False],
)
assert check_file_exists("some/path/one") is True
assert check_file_exists("some/path/two") is False
assert mock_exists.call_count == 2
assert [
call.args for call in mock_exists.call_args_list
] == [("some/path/one",), ("some/path/two",)]
# Section: Using side_effect - Callable for Multiple Calls
def test_fetch_both_endpoints_by_url(mocker: MockerFixture):
fake_responses: dict[str, MagicMock] = {}
for url, data in [
("https://api.example.com/first", {"first": "data"}),
("https://api.example.com/second", {"second": "data"}),
]:
resp = mocker.MagicMock()
resp.status_code = 200
resp.json.return_value = data
fake_responses[url] = resp
def _fake_get(url: str) -> MagicMock:
return fake_responses[url]
mocker.patch(
"dummy_functions.requests.get", side_effect=_fake_get
)
result = fetch_both_endpoints()
assert result == ({"first": "data"}, {"second": "data"})
# Section: Choosing between Mock and MagicMock
@pytest.mark.xfail(
reason="Context managers do not work with Mock", strict=True
)
def test_context_manager_with_mock(mocker: MockerFixture):
fake_cm = mocker.Mock()
fake_cm.__enter__.return_value = fake_cm
fake_cm.read.return_value = "file contents"
mock_open = mocker.patch("builtins.open")
mock_open.return_value = fake_cm
with open("somefile.txt") as f:
contents = f.read()
mock_open.assert_called_once_with("somefile.txt")
assert contents == "file contents"
def test_context_manager_with_magicmock(mocker: MockerFixture):
fake_cm = mocker.MagicMock()
fake_cm.__enter__.return_value = fake_cm
fake_cm.read.return_value = "file contents"
mock_open = mocker.patch("builtins.open")
mock_open.return_value = fake_cm
with open("somefile.txt") as f:
contents = f.read()
mock_open.assert_called_once_with("somefile.txt")
assert contents == "file contents"
@pytest.mark.parametrize(argnames, argvalues) decorator takes argument names and a list of value tuples to generate multiple test invocations.argvalues list corresponds to a separate run of the test function, with tuple elements mapped to argument names.-v shows each parametrized case as a distinct test, simplifying result interpretation.pytest.param Constructpytest.param() function wraps a set of parameter values and allows you to attach metadata to that invocation:
id: a custom label shown in the test report.marks: one or more markers (e.g., pytest.mark.xfail, pytest.mark.skip) applied only to that case.pytest.param(..., id="custom_id") to assign clear, human-readable names to individual cases.ids list passed to parametrize can specify identifiers in the order of argvalues.import pytest
def is_valid_hostname_char(char: str) -> bool:
if "a" <= char <= "z":
return True
if "0" <= char <= "9":
return True
if char == "-":
return True
return False
def check_url_status(url: str) -> tuple[int | str, str]:
if url == "https://google.com":
return (200, "OK")
if url == "https://fakesite123.org/notfound":
return (404, "HTTP_ERROR (404)")
if url == "http://httpbin.org/status/503":
return (503, "HTTP_ERROR (503)")
if url == "http://localhost:1":
return ("CONNECTION_ERROR", "CONNECTION_ERROR")
return ("UNKNOWN", "UNKNOWN")
# Section: The Problem: Duplicated Test Logic
"""
a -> True
5 -> True
- -> True
A -> False
_ -> False
"""
def test_is_valid_lower_case_a():
assert is_valid_hostname_char("a") is True
def test_is_valid_5():
assert is_valid_hostname_char("5") is True
def test_is_valid_hyphen():
assert is_valid_hostname_char("-") is True
def test_is_valid_upper_case_A():
assert is_valid_hostname_char("A") is False
def test_is_valid_underscore():
assert is_valid_hostname_char("_") is False
# Section: Solution: @pytest.mark.parametrize
@pytest.mark.parametrize(
"input_char, expected_result",
[
("a", True),
("5", True),
("-", True),
("A", False),
("_", False),
("!", False),
],
)
def test_is_valid_hostname_char(
input_char: str, expected_result: bool
):
assert is_valid_hostname_char(input_char) is expected_result
# Section: Customizing Test IDs with pytest.param construct
@pytest.mark.parametrize(
"input_char, expected_result",
[
pytest.param("a", True, id="lowercase_letter_a"),
pytest.param("z", True, id="lowercase_letter_z"),
pytest.param("0", True, id="digit_0"),
pytest.param("-", True, id="hyphen"),
pytest.param("A", False, id="uppercase_A_invalid"),
pytest.param("_", False, id="underscore_invalid"),
],
)
def test_is_valid_hostname_custom_params(
input_char: str, expected_result: bool
):
assert is_valid_hostname_char(input_char) is expected_result
@pytest.mark.parametrize(
"url_to_check, expected_status_code, expected_status_text",
[
("https://google.com", 200, "OK"),
(
"https://fakesite123.org/notfound",
404,
"HTTP_ERROR (404)",
),
(
"http://httpbin.org/status/503",
503,
"HTTP_ERROR (503)",
),
(
"http://localhost:1",
"CONNECTION_ERROR",
"CONNECTION_ERROR",
),
pytest.param(
"https://pending.retries.tests",
503,
"HTTP_ERROR (503)",
marks=(
pytest.mark.xfail(
reason="Retry logic for 503 is not yet implemented."
),
pytest.mark.api,
),
),
],
ids=[
"google_ok",
"site_not_found",
"server_error_503",
"connection_error",
"xfail_retry_case",
],
)
def test_various_url_statuses(
url_to_check: str,
expected_status_code: int,
expected_status_text: str,
):
status_code, status_text = check_url_status(url_to_check)
assert status_code == expected_status_code
assert status_text == expected_status_text
@pytest.fixture that provides a baseline environment or data for tests.@pytest.fixture@pytest.fixture above a function to mark it as a fixture that can return or yield resources.yield is setup, the yielded value goes to the test, and code after yield is teardown.function (default), class, module, or session.function-scoped fixture runs for each test function; a session-scoped fixture runs only once for the entire test session.yield rather than return.yield always runs, even if the test fails or raises an error.try...finally block, ensuring resources like temporary files or connections are properly released.conftest.py: Sharing Fixturesconftest.py file makes them available across multiple test modules without explicit imports.conftest.py in a test directory or its parent enables automatic discovery of shared fixtures.import pytest
from typing import Iterable
ManagedResource = dict[str, str]
@pytest.fixture
def managed_resource() -> Iterable[ManagedResource]:
print(" [SHARED FIXTURE]: acquiring resource lock")
resource = {"status": "lock_acquired"}
yield resource
print(" [SHARED FIXTURE]: releasing resource lock")
resource["status"] = "lock_released"@pytest.mark.<markername>) applied to tests to attach metadata.skip, skipif, xfail, and parametrize provide predefined behaviors.slow, integration, api) help categorize tests by project-specific criteria.@pytest.mark.skipskip marker unconditionally prevents a test from running, useful when a feature is disabled or under refactor.s in the summary along with the provided reason.@pytest.mark.skip(reason="...") to disable tests without removing their code.@pytest.mark.skipifskipif marker skips tests only when a specified condition evaluates to true at collection time.sys.platform != "linux" or checks for optional dependencies.@pytest.mark.xfailxfail marker marks tests expected to fail due to known bugs or unimplemented features.XFAIL and don’t cause the suite to fail; unexpected passes are reported as XPASS.slow, api, integration) help organize and categorize tests by functionality or runtime.@pytest.mark.api and @pytest.mark.integration).pytest.ini or pyproject.toml under the markers option.-m <expression> CLI option selects tests matching a marker expression.and, or, and negation with not.pytest -m slow, pytest -m "not slow", pytest -m "api and integration".skip hides failures; prefer xfail for known bugs to maintain visibility.skipif conditions reduce readability; delegate logic to helper functions when needed.import pytest
import time
try:
import some_optional_library # type: ignore
except ModuleNotFoundError:
some_optional_library = None
# Section: Skipping Tests Unconditionally: @pytest.mark.skip
@pytest.mark.skip(
reason="Skipping experimental feature until completion."
)
def test_new_experimental_feature() -> None:
assert False
# Section: Skipping Tests Conditionally: @pytest.mark.skipif
@pytest.mark.skipif(
some_optional_library is None,
reason="Requires 'some_optional_library' to be installed.",
)
def test_with_optional_dependency() -> None:
print(
f"Running tests that depends on an optional library..."
)
assert some_optional_library
# Section: Expected Failures: @pytest.mark.xfail
@pytest.mark.xfail(
reason="Bug #123: Division by zero not handled properly."
)
def test_divide_by_zero() -> None:
_division = 1 / 0
assert False
@pytest.mark.xfail # Add strict=True to make XPASS lead to a failure
def test_expected_to_fail() -> None:
assert True
# Section: Custom Markers and Registration
@pytest.mark.slow
def test_very_long_computations() -> None:
time.sleep(5)
assert True
@pytest.mark.api
@pytest.mark.smoke
def test_user_creation() -> None:
assert True
# Section: Running Tests by Marker (m option)
-v or -m on the command line every time becomes tedious.pyproject.toml under the [tool.pytest.ini_options] table.pytest without remembering flags; everyone gets the same behavior.testpaths, python_files and markers in one place.PytestUnknownMarkWarning, enable color and rich tracebacks by default.Pytest searches for settings in this order, using the first match from the current or a parent directory:
pyproject.toml under [tool.pytest.ini_options]pytest.initox.ini with a [pytest] sectionsetup.cfg under [tool:pytest]Embrace pyproject.toml as the modern hub for all your tool configurations.
pyproject.tomlpyproject.toml at your project root.[tool.pytest.ini_options] table.addopts
Defines default command-line flags that Pytest applies on every run (verbosity, reporting, color, etc.).
markers
Pre-registers custom markers with descriptions so that you can categorize tests and avoid unknown-marker warnings.
testpaths
Restricts test discovery to the listed directories, preventing Pytest from scanning other parts of the project.
python_files
Specifies filename patterns that Pytest treats as test files (e.g., test_*.py).
python_classes
Indicates class name patterns Pytest will consider when looking for test classes (e.g., classes starting with Test).
python_functions
Sets function name patterns Pytest uses to identify individual test functions (e.g., functions beginning with test_).
Other options
norecursedirs: directories to skip during discoveryminversion: enforce a minimum Pytest versionfilterwarnings: configure how warnings are handledExample of pyproject.toml
from typing import TypedDict
import re
class TextAttributes(TypedDict):
word_count: int
unique_words: set[str]
average_word_length: float
longest_word: str
def calculate_text_attributes(input_text: str) -> TextAttributes:
split_text = re.findall(r"\w+", input_text)
word_length_sum = sum(len(word) for word in split_text)
avg_word_length = (
word_length_sum / len(split_text)
if len(split_text)
else 0
)
return {
"word_count": len(split_text),
"unique_words": set(text.lower() for text in split_text),
"average_word_length": avg_word_length,
"longest_word": (
max(split_text, key=len) if split_text else ""
),
}
assert statement to declare expected conditions in tests, making test code concise and readable.assert expression evaluates to True, execution continues; if it evaluates to False, an AssertionError is raised and Pytest marks the test as failed.assert Statementassert keyword checks that an expression is truthy; if it’s falsy, Python raises AssertionError.assert expression, "message", which will be shown if the assertion fails.assert x == 5 does nothing when true, while assert x == 10, "x should be 10" raises an error with that message if the condition is false.assertassert by inspecting the expression’s values and rewriting the failure message to show variable states.False.in or not in to assert presence or absence in containers.<, >, <=, >=) to verify ordering conditions.pytest.approx)pytest.approx to compare floats within a tolerance, supporting both relative and absolute tolerances.pytest.raises)with pytest.raises(ExpectedException): as a context manager to assert that a block of code raises a specific exception.match="regex" to verify that the exception message matches a given pattern.assert; break them into multiple simpler assertions for clarity.pytest.approx for floating-point comparisons to prevent false negatives from tiny precision differences.from text_analysis import calculate_text_attributes
import pytest
# Section: The `assert` Statement
# Uncomment to play around with Python assertions
# x: int = 5
# assert x == 5 # Nothing will happen, because this is True
# assert (
# x == 10
# ), "x should be 10, but it's not!" # Raise an AssertionError
# Section: Pytest and `assert`
def test_string_equality() -> None:
expected_status = "SUCCESS"
actual_status = "success".upper()
assert actual_status == expected_status
def test_word_count() -> None:
text = "Deploying microservice to Kubernetes cluster."
text_empty = ""
assert (calculate_text_attributes(text)["word_count"]) == 5
assert (
calculate_text_attributes(text_empty)["word_count"]
) == 0
def test_unique_words() -> None:
text = "Deploying microservice to Kubernetes cluster."
text_with_duplicates = "Deploying deploying."
text_empty = ""
text_results = calculate_text_attributes(text)
text_with_duplicates_result = calculate_text_attributes(
text_with_duplicates
)
text_empty_results = calculate_text_attributes(text_empty)
assert (len(text_results["unique_words"])) == 5
assert (
len(text_with_duplicates_result["unique_words"])
) == 1
assert (len(text_empty_results["unique_words"])) == 0
def test_average_word_length() -> None:
text = "Deploying microservice to Kubernetes cluster." # 40 / 5 = 8
text_with_duplicates = "Deploying deploying." # 18 / 2 = 9
text_empty = "" # 0
text_results = calculate_text_attributes(text)
text_with_duplicates_result = calculate_text_attributes(
text_with_duplicates
)
text_empty_results = calculate_text_attributes(text_empty)
assert (text_results["average_word_length"]) == 8.0
assert (
text_with_duplicates_result["average_word_length"]
) == 9.0
assert (text_empty_results["average_word_length"]) == 0.0
def test_longest_word() -> None:
text = "Deploying microservice to Kubernetes cluster." # microservice
text_with_duplicates = "Deploying deploying." # deploying
text_empty = ""
text_results = calculate_text_attributes(text)
text_with_duplicates_result = calculate_text_attributes(
text_with_duplicates
)
text_empty_results = calculate_text_attributes(text_empty)
assert (
text_results["longest_word"].lower()
) == "microservice"
assert (
text_with_duplicates_result["longest_word"].lower()
) == "deploying"
assert (text_empty_results["longest_word"]) == ""
# Section: Pytest’s Rich Failure Output
@pytest.mark.xfail # We're marking the test as an expected failure
def test_string_mismatch() -> None:
expected = "HEllo WOrlD"
actual = "hello world"
assert expected == actual
# Section: Asserting Floating-Point Numbers (`pytest.approx`)
def test_float_with_approx() -> None:
calculated_val = 0.1 + 0.2
expected_val = 0.3
assert calculated_val == pytest.approx(expected_val) # type: ignore
# Section: Asserting Exceptions (`pytest.raises`)
def test_raises_exception() -> None:
with pytest.raises(ZeroDivisionError):
_division = 1 / 0
from typing import TypedDict
import re
class TextAttributes(TypedDict):
word_count: int
unique_words: set[str]
average_word_length: float
longest_word: str
def calculate_text_attributes(input_text: str) -> TextAttributes:
split_text = re.findall(r"\w+", input_text)
word_length_sum = sum(len(word) for word in split_text)
avg_word_length = (
word_length_sum / len(split_text)
if len(split_text)
else 0
)
return {
"word_count": len(split_text),
"unique_words": set(text.lower() for text in split_text),
"average_word_length": avg_word_length,
"longest_word": (
max(split_text, key=len) if split_text else ""
),
}
.send(), and final return values.Callable) and return a new function; using Callable[..., Any] types them broadly but loses specific signature information.TypeVar bound to Callable[..., Any] and use it for both the decorator’s input and output types.*args: Any, **kwargs: Any -> Any, while TypeVar ensures the decorated function’s overall type remains correct.Generator[YieldType, SendType, ReturnType] to specify a generator’s yield type, the type accepted by .send(), and its return type on completion.send(), set SendType to None; if it has no explicit return, set ReturnType to None.count_up generator is typed as Generator[int, None, str], yielding integers and returning a string message.accumulate_and_send generator is typed as Generator[float, float, None], yielding a running total, accepting floats via send(), and returning nothing.Iterable[T] to accept any iterable of T (lists, tuples, generators).Iterator[T] when a function specifically expects an iterator object supporting __next__().from typing import (
Callable,
Any,
TypeVar,
ParamSpec,
Generator,
Iterable,
)
import functools
# Section: Typing Decorators (simple_logging_decorator)
def simple_logging_decorator(
func: Callable[..., Any],
) -> Callable[..., Any]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
print(f"LOG: Calling {func.__name__}")
result = func(*args, **kwargs)
print(f"LOG: {func.__name__} returned {result}")
return result
return wrapper
@simple_logging_decorator
def add(x: int, y: int) -> int:
return x + y
result_add = add(3, 5)
# Section: Typing Decorators (better_logging_decorator with TypeVar)
P = ParamSpec("P")
R = TypeVar("R")
def better_logging_decorator(
func: Callable[P, R],
) -> Callable[P, R]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
print(f"LOG: Calling {func.__name__}")
result = func(*args, **kwargs)
print(f"LOG: {func.__name__} returned {result}")
return result
return wrapper
@better_logging_decorator
def subtract(x: int, y: int) -> int:
return x - y
result_subtract = subtract(3, 5)
# Section: Typing Generators
def count_up_to(limit: int) -> Generator[int, None, str]:
for i in range(limit):
yield i
return "Counting complete!"
def accumulate_and_send() -> (
Generator[float, float | None, None]
):
total = 0.0
try:
while True:
sent = yield total
if sent:
total += sent
except GeneratorExit:
pass
test_accumulate = accumulate_and_send()
next(test_accumulate)
print(test_accumulate.send(1.0))
print(next(test_accumulate))
print(test_accumulate.send(2.0))
print(test_accumulate.send(3.0))
print(next(test_accumulate))
# Section: Iterable & Iterator
def process_items(items: Iterable[str]) -> list[str]:
return [item.upper() for item in items]
print(process_items(["a", "b"]))
print(process_items(("a", "b")))
print(process_items({"a", "b"}))
print(process_items({"a": "b", "hello": "world"}))
Any.typing module’s TypeVar and Generic primitives unlock this capability.Any sacrifices type information, so tools cannot guarantee correct usage of returned values.str for a list[str] and int for a list[int], not just Any.T = TypeVar('T') declares a placeholder type variable T that can stand for any type.def get_first_item_generic(data: List[T]) -> Optional[T]: returns an element of the same type as the list elements.T from each call site, preserving specific return types like Optional[str] or Optional[int].NumberType = TypeVar('NumberType', int, float).def add_generic_numbers(x: NumberType, y: NumberType) -> NumberType: then only accept int or float and return that same type.NumberType = TypeVar('NumberType', bound=Superclass).def add_generic_numbers(x: NumberType, y: NumberType) -> NumberType: then accept any subclass of Superclass, and they can be different subclasses for each argument.Generic[T] to define a class parameterized by a type variable T.SimpleStack[T] can push, pop, and peek items of type T, and MyPy will enforce that only T instances are used.T in its methods but does not inherit from Generic[T] is not recognized as generic by MyPy.TypeVar('T') can degrade type safety when operations require certain capabilities—use bounds or explicit type lists when appropriate.from typing import Optional, TypeVar, Generic
# Section: Defining a generic function to get the first item of a list
T = TypeVar("T")
def get_first_item(
input_list: list[T],
) -> Optional[T]:
if input_list:
return input_list[0]
return None
first_number = get_first_item([1, 2, 3])
first_str = get_first_item(["abc", "def"])
first_mixed_list = get_first_item(["abc", "def", 1, 2, 3])
# Section: Constrained TypeVar for numeric addition
NumberType = TypeVar("NumberType", int, float)
def add_generic_numbers(
x: NumberType, y: NumberType
) -> NumberType:
return x + y
sum_int = add_generic_numbers(3, 5.0)
# Section: Bounded TypeVar with deployed filter for DevOps resources
class CloudResource:
def __init__(self, name: str, cpu_usage: float) -> None:
self.name = name
self.cpu_usage = cpu_usage
self.deployed: bool = False
def deploy(self) -> None:
print(f"Deploying {self.name}")
self.deployed = True
class VirtualMachine(CloudResource):
def reboot(self) -> None:
print(f"Rebooting VM {self.name}")
class DockerContainer(CloudResource):
def restart(self) -> None:
print(f"Restarting container {self.name}")
ResourceType = TypeVar("ResourceType", bound=CloudResource)
def filter_deployed(
resources: list[ResourceType],
) -> list[ResourceType]:
return [
resource for resource in resources if resource.deployed
]
vm1 = VirtualMachine("vm-01", cpu_usage=65.0)
vm2 = VirtualMachine("vm-02", cpu_usage=45.0)
container1 = DockerContainer("api-service", cpu_usage=85.0)
container2 = DockerContainer("worker", cpu_usage=55.0)
vm1.deploy()
container1.deploy()
all_resources = [vm1, vm2, container1, container2]
deployed_resources = filter_deployed(all_resources)
# Section: Generic class SimpleStack
G = TypeVar("G")
class SimpleStack(Generic[G]):
def __init__(self) -> None:
self._items: list[G] = []
def push(self, item: G) -> None:
self._items.append(item)
def pop(self) -> G:
if self.is_empty():
raise IndexError("Stack is empty!")
return self._items.pop()
def peek(self) -> Optional[G]:
if self.is_empty():
return None
return self._items[-1]
def is_empty(self) -> bool:
return not self._items
str_stack = SimpleStack[str](http://)
str_stack.push("str")
int_stack = SimpleStack[int](http://)
int_stack.push(12)
self is implicitly the class type; annotate other parameters and return types normally.typing.Self for methods that return the instance, for example def clone(self) -> Self:.from __future__ import annotations defers evaluation of all annotations, simplifying forward references.from __future__ import annotations
from typing import Self, Optional
# Section: Classes as Type Hints
class Server:
def __init__(
self,
hostname: str,
ip_address: str,
os_type: str = "Linux",
):
self.hostname: str = hostname
self.ip_address: str = ip_address
self.os_type: str = os_type
self.is_online: bool = False
def connect(self) -> None:
print(
f"Connecting to {self.hostname} (IP address: {self.ip_address})"
)
self.is_online = True
print(f"{self.hostname} is online.")
def get_status(self) -> str:
return "online" if self.is_online else "offline"
def deploy_app_to_server(
target_server: Server, app_name: str
) -> bool:
print(
f"Deploying {app_name} to server: {target_server.hostname}"
)
if not target_server.is_online:
target_server.connect()
print(
f"Deployment of {app_name} to {target_server.hostname} successful."
)
return True
web_server = Server(
hostname="web01.dev.local", ip_address="10.0.1.10"
)
db_server = Server(
hostname="db01.dev.local", ip_address="10.0.2.20"
)
deploy_app_to_server(web_server, "FrontendApp")
deploy_app_to_server(db_server, "UserDBApi")
# Section: Hinting Methods Within a Class
class Calculator:
def __init__(self, initial_value: int | float = 0):
self.total: int | float = initial_value
def add(self, value: int | float) -> Self:
self.total += value
return self
def subtract(self, value: int | float) -> Self:
self.total -= value
return self
def multiply_by(self, value: int | float) -> Self:
self.total *= value
return self
def divide_by(self, value: int | float) -> Self:
self.total /= value
return self
def get_total(self) -> int | float:
return self.total
my_calc = Calculator(1)
print(my_calc.add(2).subtract(1).multiply_by(10).get_total())
# Section: Forward References (Strings)
class Employee:
def __init__(
self, name: str, manager: Optional[Employee] = None
) -> None:
self.name: str = name
self.manager: Optional[Employee] = manager
self.reports: list[Employee] = []
def add_report(self, report: Employee) -> None:
self.reports.append(report)
ceo = Employee("ceo")
manager1 = Employee("Alice", ceo)
ceo.add_report(manager1)
def process_user_data(user: dict) -> bool: makes it clear that the function expects a dict and returns a bool.variable_name: type = value. This syntax (variable annotations) was introduced in Python 3.6 (PEP 526).config_path: str = "/etc/app.conf" indicates that config_path is intended to be a string.param_name: param_type, and the return type is specified after -> before the colon.def get_server_status(hostname: str, port: int) -> str: declares that the function takes a str and an int, and returns a str.int, float, bool, str, and bytes are directly usable in annotations.list, tuple, set, and dict. For more precise element types:
list[int], dict[str, int].typing module: from typing import List, Dict and use List[int], Dict[str, int].None is used for functions that do not return a meaningful value (e.g., -> None).Optional, Union, and others will be covered when exploring the typing module in a later lecture.process_id("user-123") on a function annotated as def process_id(user_id: int) -> None: runs without a hint-triggered error, though passing a string where an integer is expected may lead to a TypeError later if arithmetic is attempted.Any for truly dynamic values.typing imports: When using List[int], Optional[str], etc., remember to import them from the typing module (unless you rely on built-in generics in Python 3.9+).# Section: Basic Type Hint Syntax - Variable Annotations
config_path: str = "/etc/app.conf"
retry_count: int = 3
is_enabled: bool = bool(1)
servers: list[str] = ["web01", "web02"]
settings: dict[str, int | str] = {"port": 8080, "user": "admin"}
# Section: Basic Type Hint Syntax - Function Argument and Return Type Annotations
def get_server_status(hostname: str, port: int) -> str:
print(f"Checking {hostname}:{port}")
if port == 80:
return "Online"
else:
return "Unknown"
# Section: Python Remains Dynamically Typed
def process_id(user_id: int) -> None:
print(
f"Processing user ID: {user_id} (type: {type(user_id)})"
)
# Demonstration of dynamic typing
process_id(1234)
# process_id("user-1234") # Uncommenting will lead to a static type checking error.
typing module provides specialized type constructors to precisely describe the contents of collections (list, dict, tuple, set) and other complex scenarios.typing Modulelist[int], dict[str, str], tuple[int, ...], set[str], frozenset[int]) are available via PEP 585, deprecating typing.List etc. for these cases.typing, for example: List, Dict, Tuple, Set, FrozenSet, Optional, Union, Any.typing remains necessary for compatibility with older versions (Python 3.7/3.8) and for constructs like Optional, Union, Literal, and TypedDict.list[X] (or List[X] in Python < 3.9) to indicate a list whose elements are of type X.list[str]) or integers (list[int]), enabling static checkers to catch mismatches.dict[K, V] (or Dict[K, V] in Python < 3.9) to specify a dictionary with keys of type K and values of type V.dict[int, list[str]], to model complex structures like mapping user IDs to role lists.from typing import TypedDict, NotRequired
class User(TypedDict):
id: int
name: str
email: str
phone: NotRequired[str]
user: User = {
"id": 123,
"name": "Alice",
"email": "alice@example.com",
"phone": "+123456789",
}
print(f"User data: {user.get("email")}")
tuple[T1, T2, ...] (or Tuple[T1, T2, ...] in Python < 3.9).tuple[T, ...] (or Tuple[T, ...]), though lists are often more natural for that use case.set[X] (or Set[X] in Python < 3.9) to indicate a set containing elements of type X.in) will compare values of the declared type.frozenset[X] (or FrozenSet[X] in Python < 3.9).Union[...] when a value may be exactly one of several types (excluding None unless explicitly included).int | str instead of Union[int, str].Optional[X] is shorthand for Union[X, None], indicating a value may be of type X or None.Optional value without first checking for None.Any disables type checking for the annotated part, useful during gradual typing of legacy code or when truly dynamic types are needed.list[int] only works on Python 3.9+; use typing.List[int] for Python 3.7/3.8 compatibility.def func(arg: Optional[str] = None) clearly allows None as a default, whereas def func(arg: str = None) may confuse static checkers.Any: Reserving Any for truly dynamic cases preserves the value of static checking elsewhere in your code.from typing import Optional, Any
# Section: Typing Lists
hostnames: list[str] = ["web01.example.com", "db01.example.com"]
open_ports: list[int] = [80, 443, 22]
def process_hostnames(hosts: list[str]) -> None:
for host in hosts:
print(f"Processing host: {host.upper()}")
process_hostnames(hostnames)
# process_hostnames(open_ports) # Uncommenting will lead to type error
# Section: Typing Dictionaries
server_config: dict[str, str] = {
"hostname": "app01.prod",
"ip_address": "10.0.5.20",
"os_type": "Linux",
}
user_roles: dict[str, list[str]] = {
"user-123": ["admin", "editor"],
"user-456": ["dev", "viewer"],
}
# Section: Typing Tuples
server_status: tuple[str, int, bool] = (
"api.example.com",
443,
True,
)
ip_parts: tuple[int, ...] = (192, 168, 1, 100)
# Section: Typing Sets
admin_users: set[str] = {"alice", "bob", "charlie"}
def is_admin(username: str, admins: set[str]) -> bool:
return username in admins
# Section: Union[X, Y, ...] for Multiple Possible Types
identifier: str | int = "abcde-1234"
identifier = 1234
def process_mixed_data(data: list[int | str]) -> None:
for item in data:
if isinstance(item, str):
print(f"Processing string: {item.upper()}")
else:
print(f"Processing int: {item * 2}")
# Section: Optional[X] for Values That Can Be None
def find_user(user_id: str) -> Optional[dict[str, str]]:
if user_id == "123":
return {
"id": "123",
"name": "Admin user",
"email": "admin@example.com",
}
return None
found_user = find_user("123")
if found_user:
print(f"Found user: {found_user["name"]}")
# Section: Any for Unrestricted Types
def print_anything(item: Any) -> None:
print(f"Item: {item}, type: {type(item)}")
print_anything(1)
print_anything("hello")
requests may wait indefinitely for a response, which is risky in automation.timeout parameter with a single value for both connect and read, or a tuple (connect, read) for fine-grained control.ConnectTimeout is raised if the connection can’t be established in time; a ReadTimeout is raised if data stops arriving within the read timeout.HTTPBIN_ENDPOINT = "https://httpbin.org"
import requests
import time
delay_url = f"{HTTPBIN_ENDPOINT}/delay/5" # Simulate a 5-second delay
start = time.perf_counter()
try:
res = requests.get(delay_url, timeout=2)
print(f"Completed in {time.perf_counter() - start:.2f}s, status {response.status_code}")
except (
requests.exceptions.ConnectTimeout,
requests.exceptions.ReadTimeout
) as timeout_err:
print(f"Timeout after {time.perf_counter() - start:.2f}s: {timeout_err}")
import requests
import time
flaky_url = f"{HTTPBIN_ENDPOINT}/status/200,500,503"
max_retries = 3
delay = 2
for attempt in range(1, max_retries + 1):
print(f"Attempt {attempt}/{max_retries}...")
try:
res = requests.get(flaky_url, timeout=10)
res.raise_for_status()
print(f"Succeeded with status {res.status_code}")
break
except requests.exceptions.HTTPError as err:
if err.response.status_code < 500:
print(f"Failed with client error code {err.response.status_code}. Skipping retry.")
break
else:
print(f"Failed with server error code {err.response.status_code}.")
if attempt < max_retries:
print(f"Waiting {delay}s before retry...")
time.sleep(delay)
else:
print(f"All {max_retries} attempts failed!")
import requests
import time
import random
def get_with_backoff(url, max_retries=3):
delay=1
for attempt in range(1, max_retries + 1):
print(f"Attempt {attempt}/{max_retries}...")
try:
res = requests.get(url, timeout=10)
res.raise_for_status()
print(f"Succeeded with status {res.status_code}")
return res
except requests.exceptions.HTTPError as err:
if err.response.status_code < 500:
print(f"Failed with client error code {err.response.status_code}. Skipping retry.")
raise RuntimeError(f"Client error! Please review request.")
else:
jitter = random.uniform(-0.1 * delay, 0.1 * delay)
# delay = 1 -> jitter [-0.1, 0.1] -> 0.9 and 1.1s
# delay = 2 -> jitter [-0.2, 0.2] -> 1.8 and 2.2s
# delay = 4 -> jitter [-0.4, 0.4] -> 3.6 and 4.4s
wait = min(delay * 2, 30) + jitter
print(f" Failed with server error code {err.response.status_code}. Retrying in {wait:.2f}s")
time.sleep(wait)
delay = min(delay * 2, 30)
raise RuntimeError(f"All retries to query {url} failed!")
try:
res = get_with_backoff(
f"{HTTPBIN_ENDPOINT}/status/503",
max_retries=4
)
except RuntimeError as e:
print(e)
timeout.GITHUB_ENDPOINT = "https://api.github.com"
HTTPBIN_ENDPOINT = "https://httpbin.org"
import requests
urls = {
"public_endpoint": f"{GITHUB_ENDPOINT}/zen",
"protected_endpoint": f"{GITHUB_ENDPOINT}/user",
}
for description, url in urls.items():
res = requests.get(url, timeout=5)
print(f"{description} ({url}) : {res.status_code}")
print(res.text[:200])
Authorization header.requests accepts an auth=(username, password) tuple and handles encoding automatically.401 Unauthorized when credentials are missing or incorrect.import requests
import json
url = f"{HTTPBIN_ENDPOINT}/basic-auth/myuser/myotherpwd"
try:
res = requests.get(url, auth=("myuser", "mypasswd"), timeout=10)
res.raise_for_status()
print(f"Status code: {res.status_code}")
print("Response JSON:")
print(json.dumps(res.json()))
except requests.exceptions.HTTPError as err:
print(err)
Authorization header.import requests
import os
from dotenv import load_dotenv
load_dotenv(override=True)
token = os.getenv("GH_PAT", "")
print(f"Token: {token[:15]}")
urls = {
"public_endpoint": f"{GITHUB_ENDPOINT}/zen",
"protected_endpoint": f"{GITHUB_ENDPOINT}/user",
}
for description, url in urls.items():
try:
headers = {
"Authorization": f"Bearer {token}"
}
res = requests.get(url, headers=headers, timeout=10)
res.raise_for_status()
print(f"Status code: {res.status_code}")
print(f"Authenticated user: {res.json().get("login")}")
except requests.exceptions.JSONDecodeError as err:
print(f"Invalid JSON in response body. Defaulting to text:")
print(res.text[:200])
except requests.exceptions.HTTPError as err:
print(err)
Bearer vs token) causes 401/403 errors. Follow API docs.200 OK means success, while codes like 404 Not Found or 500 Internal Server Error indicate different failure modes.response.ok, raise errors automatically, and inspect error details for troubleshooting.200 OK, 201 Created, 301 Moved Permanently, 404 Not Found, and 500 Internal Server Error.response.status_coderequests call, the integer response.status_code tells you the exact HTTP code returned.if resp.status_code == 404:) to implement custom logic based on the code.GITHUB_ENDPOINT = "https://api.github.com"
HTTPBIN_ENDPOINT = "https://httpbin.org"
import requests
urls = {
"ok": f"{GITHUB_ENDPOINT}/zen",
"not_found": f"{GITHUB_ENDPOINT}/nonexistentendpoint"
}
for description, url in urls.items():
response = requests.get(url, timeout=5)
print(f"{description}: status {response.status_code}")
response.okresponse.ok is True for any status code below 400 (1xx, 2xx, 3xx) and False for 4xx/5xx errors.import requests
urls = {
"ok": f"{GITHUB_ENDPOINT}/zen",
"not_found": f"{GITHUB_ENDPOINT}/nonexistentendpoint"
}
for description, url in urls.items():
response = requests.get(url, timeout=5)
print(f"{description}: ok? {"Yes" if response.ok else f"No. Failed with status {response.status_code}"}")
raise_for_status()response.raise_for_status() will do nothing on 1xx, 2xx and 3xx codes but raise an HTTPError on 4xx/5xx.response in its response attribute, letting you inspect headers and body.import requests
import json
urls = {
"ok": f"{GITHUB_ENDPOINT}/zen",
"not_found": f"{GITHUB_ENDPOINT}/nonexistentendpoint"
}
for url in urls.values():
print(f"Requesting: {url}")
try:
res = requests.get(url, timeout=5)
res.raise_for_status()
print(" Success!")
except requests.exceptions.HTTPError as err:
print(f" HTTPError: {err} (status {err.response.status_code})")
try:
details = err.response.json()
print(" Error details:")
print(json.dumps(details, indent=2))
except ValueError:
print(f" Non-JSON response body: {err.response.text[:100]}")
ok or raise_for_status().except Exception: hides HTTP errors. Catch HTTPError specifically.response.text or response.json().requests library simplifies HTTP interactions by abstracting raw HTTP details, making it ideal for DevOps automation tasks.requests, performing GET and POST requests, inspecting response data, and customizing requests with parameters and headers.requests library is a third-party package and must be installed in your active virtual environment.pip install requests==2.32.2 to add it to your project (pinning the version here so that we all work with the same version, but in other projects you can omit the version to install the latest), and consider pinning its version in requirements.txt.GITHUB_ENDPOINT = "https://api.github.com"
requests.get()url, optional params for query strings, headers for custom HTTP headers, and timeout to avoid hanging requests.Response object provides .status_code, .headers, .text, .content, and .json(), plus .raise_for_status() to handle HTTP errors.import requests
import json
response = requests.get(GITHUB_ENDPOINT, timeout=10)
print(f"Status code: {response.status_code}")
print(f"Content-Type: {response.headers.get("Content-Type")}")
"""
# Commenting out for brevity, but leaving for documentation
print(".text attribute:")
print(response.text)
print("\n")
print(".content attribute:")
print(response.content)
print("\n")
print(".json() method:")
print(response.json())
"""
data = response.json()
print("Available endpoints:")
print(json.dumps(data, indent=2))
paramsparams argument, and requests handles URL-encoding automatically.response.url to confirm your parameters were applied correctly.import requests
import json
search_url = f"{GITHUB_ENDPOINT}/search/repositories"
query_params = {
"q": "python devops",
"sort": "stars",
"order": "desc",
"per_page": 5
}
response = requests.get(search_url, params=query_params, timeout=10)
response.raise_for_status()
print(f"Requested URL: {response.url}")
results = response.json()
print(f"Found {results.get("total_count")} repositories. Top 5:")
for repo in results.get("items", []):
print(f"- {repo["name"]} (Stars: {repo["stargazers_count"]})")
print(json.dumps(results.get("items", [])[0], indent=2))
requests.post()requests.post() to send data to a server, choosing between data= for form-encoded bodies or json= for JSON payloads.json= automatically serializes it and sets Content-Type: application/json..status_code, .json(), and error handling.import requests
import json
post_echo_url = "https://httpbin.org/post"
payload = {
"script_name": "devops_automation",
"action": "trigger_deployment",
"environment": "staging",
"version": "v1.5.0"
}
response = requests.post(post_echo_url, json=payload, timeout=10)
response.raise_for_status()
print(json.dumps(response.json(), indent=2))
timeout value.response.raise_for_status().data= instead of json= sends form-encoded data, which may be rejected by modern APIs expecting JSON.python-dotenv) and pass them in headers.