Custom Header Manipulation in aiohttp

In this tutorial, you’ll learn various methods for custom header handling in aiohttp, from basic operations to advanced cases.

You’ll learn how to set, get, and modify headers, work with default headers, handle case sensitivity, generate dynamic headers, and more.

 

 

Set and Get Headers

Set Custom Headers to Requests

To add custom headers to your aiohttp requests, use the headers parameter in the request methods.

import aiohttp
import asyncio
async def fetch_with_custom_headers(url):
    headers = {
        'X-Custom-Header': 'MyValue',
        'Authorization': 'Bearer token123'
    }
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            return await response.text()
url = 'https://httpbin.org/headers'
response_text = asyncio.run(fetch_with_custom_headers(url))
print(response_text)

Output:

{
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Authorization": "Bearer token123", 
    "Host": "httpbin.org", 
    "User-Agent": "Python/3.11 aiohttp/3.9.5", 
    "X-Amzn-Trace-Id": "Root=1-66d439b9-0307473e1553866809891e9e", 
    "X-Custom-Header": "MyValue"
  }
}

The output shows that the custom headers ‘X-Custom-Header’ and ‘Authorization’ were successfully added to the request.

The server echoed back all received headers including the custom ones we set.

Get Response Headers

To access headers from the server’s response, use the headers attribute of the response object.

async def fetch_and_print_headers(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            print(dict(response.headers))
url = 'https://httpbin.org/get'
asyncio.run(fetch_and_print_headers(url))

Output:

{'Date': 'Sun, 01 Sep 2024 09:57:15 GMT', 'Content-Type': 'application/json', 'Content-Length': '312', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true'}

The output displays all headers returned by the server.

This includes important information like the content type, server details, and access control settings.

Update headers in existing requests

You can update headers for each request, even within the same session by providing new headers to the request method.

async def update_headers_in_session(url):
    async with aiohttp.ClientSession() as session:
        headers1 = {'X-Request-ID': '001'}
        async with session.get(url, headers=headers1) as response1:
            print("First request headers:", dict(response1.request_info.headers))

        headers2 = {'X-Request-ID': '002', 'X-Custom': 'NewValue'}
        async with session.get(url, headers=headers2) as response2:
            print("Second request headers:", dict(response2.request_info.headers))
url = 'https://httpbin.org/headers'
asyncio.run(update_headers_in_session(url))

Output:

First request headers: {'Host': 'httpbin.org', 'X-Request-ID': '001', 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'User-Agent': 'Python/3.11 aiohttp/3.9.5'}
Second request headers: {'Host': 'httpbin.org', 'X-Request-ID': '002', 'X-Custom': 'NewValue', 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'User-Agent': 'Python/3.11 aiohttp/3.9.5'}

The first request includes the ‘X-Request-ID’ header with value ‘001’, while the second request updates this header to ‘002’ and adds a new ‘X-Custom’ header.

 

Default Headers

aiohttp sets some default headers automatically.

You can inspect these by looking at the request headers.

async def check_default_headers(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            print(dict(response.request_info.headers))
url = 'https://httpbin.org/get'
asyncio.run(check_default_headers(url))

Output:

{'Host': 'httpbin.org', 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'User-Agent': 'Python/3.11 aiohttp/3.9.5'}

Modify or remove default headers

To modify or remove default headers, you can create a custom ClientSession with updated default headers.

async def modify_default_headers(url):
    default_headers = {
        'User-Agent': 'MyCustomAgent/1.0',
        'Accept': 'application/json'
    }
    async with aiohttp.ClientSession(headers=default_headers) as session:
        async with session.get(url) as response:
            print(dict(response.request_info.headers))
url = 'https://httpbin.org/get'
asyncio.run(modify_default_headers(url))

Output:

{'Host': 'httpbin.org', 'User-Agent': 'MyCustomAgent/1.0', 'Accept': 'application/json', 'Accept-Encoding': 'gzip, deflate'}

The output shows that the default ‘User-Agent’ has been replaced with ‘MyCustomAgent/1.0’, and the ‘Accept’ header has been set to ‘application/json’. Other default headers remain unchanged.

Set a custom User-Agent

You can set a custom User-Agent by including it in the headers or session defaults.

async def set_custom_user_agent(url):
    headers = {'User-Agent': 'MyApp/2.0 (Custom Python Client)'}
    async with aiohttp.ClientSession(headers=headers) as session:
        async with session.get(url) as response:
            print(response.request_info.headers['User-Agent'])
url = 'https://httpbin.org/user-agent'
asyncio.run(set_custom_user_agent(url))

Output:

MyApp/2.0 (Custom Python Client)

 

Header Case Sensitivity

aiohttp handles headers in a case-insensitive way, which aligns with the HTTP specification.

async def test_header_case_sensitivity(url):
    headers = {'Content-Type': 'application/json', 'X-Custom-Header': 'Value'}
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            print("Original:", response.headers['Content-Type'])
            print("Lowercase:", response.headers['content-type'])
            print("Mixed l:", response.headers['SERVER'])
            print("Mixed 2:", response.headers['SeRVer'])
url = 'https://httpbin.org/headers'
asyncio.run(test_header_case_sensitivity(url))

Output:

Original: application/json
Lowercase: application/json
Mixed l): gunicorn/19.9.0
Mixed 2: gunicorn/19.9.0

You can access header values using any combination of upper and lowercase letters, and you’ll still retrieve the correct value.

 

Dynamic Header Generation

Generate headers dynamically based on runtime conditions or user input to customize each request.

import time
async def dynamic_headers(url, user_id):
    def generate_headers(user_id):
        timestamp = int(time.time())
        return {
            'X-User-ID': str(user_id),
            'X-Timestamp': str(timestamp),
            'X-Session-ID': f"{user_id}-{timestamp}"
        }

    async with aiohttp.ClientSession() as session:
        headers = generate_headers(user_id)
        async with session.get(url, headers=headers) as response:
            print(await response.json())
url = 'https://httpbin.org/headers'
asyncio.run(dynamic_headers(url, 12345))

Output:

{'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.11 aiohttp/3.9.5', 'X-Amzn-Trace-Id': 'Root=1-66d45a26-56c4f65e683d5c1943e090a0', 'X-Session-Id': '12345-1725192744', 'X-Timestamp': '1725192744', 'X-User-Id': '12345'}}

The output shows dynamically generated headers including a user ID, timestamp, and a session ID composed of both.

This method allows for unique identification of each request based on runtime conditions.

 

Header Serialization

For complex data structures, you can serialize them into a string format (like JSON) to include in headers.

import json
async def serialize_header_data(url):
    complex_data = {
        'user': {'id': 1, 'name': 'John Doe'},
        'permissions': ['read', 'write']
    }
    headers = {'X-Metadata': json.dumps(complex_data)}
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            print(await response.json())
url = 'https://httpbin.org/headers'
asyncio.run(serialize_header_data(url))

Output:

{'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.11 aiohttp/3.9.5', 'X-Amzn-Trace-Id': 'Root=1-66d45a6e-6c38a85934608e4f3365eee6', 'X-Metadata': '{"user": {"id": 1, "name": "John Doe"}, "permissions": ["read", "write"]}'}}

The output shows that the complex data structure was successfully serialized into a JSON string and included in the ‘X-Metadata’ header.

This allows you to send structured data as part of your request headers.

 

Deserialization of Response Headers

When receiving complex headers from a server response, you’ll need to deserialize them back into usable data structures.

async def deserialize_header_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            metadata = response.headers.get('X-Metadata')
            if metadata:
                data = json.loads(metadata)
                print("Deserialized metadata:", data)
                print("User name:", data['user']['name'])
                print("Permissions:", ', '.join(data['permissions']))
url = 'https://httpbin.org/response-headers?X-Metadata={"user": {"id": 1, "name": "John Doe"}, "permissions": ["read", "write"]}'
asyncio.run(deserialize_header_data(url))

Output:

Deserialized metadata: {'user': {'id': 1, 'name': 'John Doe'}, 'permissions': ['read', 'write']}
User name: John Doe
Permissions: read, write

The JSON string is parsed back into a Python dictionary, allowing easy access to nested data like the user’s name and permissions.

 

Conditional Headers

You can use conditional headers to make requests that depend on the state of the resource on the server.

async def conditional_request(url, etag):
    headers = {'If-None-Match': etag}
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            if response.status == 304:
                print("Resource not modified")
            else:
                print("Resource modified, new content:", await response.text())
            print("New ETag:", response.headers.get('ETag'))
url = 'https://httpbin.org/etag/test'
etag = '"test"'  # This should match the resource's current ETag
asyncio.run(conditional_request(url, etag))

Output:

Resource not modified
New ETag: "test"

The output indicates that the resource was not modified since the provided ETag still matches.

This shows how conditional headers can be used to avoid unnecessary data transfer when the resource hasn’t changed.