Create a .env file with all your Chainstack endpoints with Python.
Create a .env file with all your Chainstack endpoints with Python.
Learn how to automatically generate a .env file listing all your Chainstack endpoints with Python.
import requests
import os
import logging
from dotenv import load_dotenv
from web3 import Web3
from typing import Optional, Dict, Any
# Load environment variables
load_dotenv()
# Constants
CHAINSTACK_API_KEY = os.getenv('CHAINSTACK_API_KEY')
OUTPUT_FILE_NAME = 'rpc.env'
# Initialize logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def fetch_chainstack_data(api_key: str) -> Optional[Dict[str, Any]]:
"""Fetch data from Chainstack API."""
url = "https://api.chainstack.com/v1/nodes/"
headers = {
"accept": "application/json",
"authorization": f"Bearer {api_key}"
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
logging.info(f"Fetched {len(data.get('results', []))} items from Chainstack.")
return data
except requests.RequestException as e:
logging.error(f"Failed to fetch data from Chainstack: {e}")
return None
def process_chainstack_item(item: Dict[str, Any]) -> Dict[str, str]:
"""Process a single item from Chainstack data."""
logging.debug(f"Processing item: {item['name']} with ID {item['id']}")
return {
'id': item['id'],
'name': item['name'],
'details': item['details'],
'http_endpoint': item['details'].get('https_endpoint'),
'auth_key': item['details'].get('auth_key'),
'configuration': item['configuration'],
'client': item['configuration'].get('client')
}
def connect_to_web3(reconstructed_endpoint: str) -> bool:
"""Connect to a Web3 endpoint."""
logging.debug(f"Attempting to connect to Web3 endpoint: {reconstructed_endpoint}")
try:
w3 = Web3(Web3.HTTPProvider(reconstructed_endpoint))
if w3.is_connected():
chain_id = w3.eth.chain_id
logging.info(f"Connected to {reconstructed_endpoint} with chain ID: {chain_id}")
return True
else:
logging.warning(f"Failed to connect to {reconstructed_endpoint}")
except Exception as e:
logging.error(f"An error occurred while connecting to {reconstructed_endpoint}: {e}")
return False
def sanitize_name(name: str) -> str:
"""Sanitize the endpoint name for use as an environment variable key."""
return name.replace(" ", "_").replace("-", "_").replace("/", "_").upper()
def create_env_file(endpoint_info_dict: Dict[str, Dict[str, str]], filename: str = OUTPUT_FILE_NAME) -> None:
"""Create a .env file from the endpoint info dictionary."""
logging.info(f"Preparing to write {len(endpoint_info_dict)} endpoints to .env file.")
with open(filename, 'w') as file:
for endpoint, info in endpoint_info_dict.items():
sanitized_name = sanitize_name(info['name'])
file.write(f'{sanitized_name}_URL="{endpoint}"\n')
logging.info(f".env file created successfully at {filename}.")
def main() -> None:
"""Main function to orchestrate the process."""
logging.info("Starting main process.")
if not CHAINSTACK_API_KEY:
logging.error("Chainstack API key not found.")
return
json_data = fetch_chainstack_data(CHAINSTACK_API_KEY)
if not json_data:
return
endpoint_info_dict = {}
for item in json_data.get('results', []):
data = process_chainstack_item(item)
reconstructed_endpoint = f"{data['http_endpoint']}/{data['auth_key']}"
if connect_to_web3(reconstructed_endpoint):
endpoint_info_dict[reconstructed_endpoint] = {'name': data['name']}
if endpoint_info_dict:
create_env_file(endpoint_info_dict)
else:
logging.info("No endpoint information to write to .env file.")
logging.info("Main process completed.")
if __name__ == "__main__":
main()
2024-01-09 17:50:27,635 - INFO - Starting main process.
2024-01-09 17:50:29,556 - INFO - Fetched 9 items from Chainstack.
2024-01-09 17:50:30,707 - INFO - Connected to https://nd-422-757-666.p2pify.com/ with chain ID: 1
2024-01-09 17:50:33,324 - INFO - Connected to https://nd-828-700-214.p2pify.com/ with chain ID: 137
2024-01-09 17:50:34,953 - WARNING - Failed to connect to https://nd-418-459-126.p2pify.com/
2024-01-09 17:50:36,296 - INFO - Connected to https://nd-000-364-211.p2pify.com/ with chain ID: 42161
2024-01-09 17:50:38,214 - WARNING - Failed to connect to https://nd-326-444-187.p2pify.com/
2024-01-09 17:50:40,403 - INFO - Connected to https://nd-500-249-268.p2pify.com/ with chain ID: 100
2024-01-09 17:50:41,948 - INFO - Connected to https://nd-363-550-219.p2pify.com/ with chain ID: 1101
2024-01-09 17:50:42,927 - INFO - Connected to https://nd-954-882-037.p2pify.com/ with chain ID: 42161
2024-01-09 17:50:43,607 - WARNING - Failed to connect to https://starknet-mainnet.core.chainstack.com/
2024-01-09 17:50:43,607 - INFO - Preparing to write 6 endpoints to .env file.
2024-01-09 17:50:43,608 - INFO - .env file created successfully at rpc.env.
2024-01-09 17:50:43,608 - INFO - Main process completed.
Chainstack provides a robust API that enables you to retrieve data about your projects and endpoints efficiently. This short tutorial will use the list all nodes API.
This API allows you to seamlessly extract a comprehensive list of nodes associated with your account. Additionally, we will demonstrate how to automate the creation of a .env file with the fetched data.
Note that this script is designed for EVM compatible chains.
You will need a Chainstack account with an API and Python
To successfully set up your project environment, follow these steps:
1
Create a New Python Virtual Environment
This isolated environment ensures that your project dependencies do not interfere with the global Python setup. Use the following command in your terminal:
Bash
python3 -m venv get_endpoints
This command creates a new virtual environment named get_endpoints.
2
Activate the Virtual Environment
To activate the environment, run:
Bash
source get_endpoints/bin/activate
After activation, you will notice that the terminal prompt is prefixed with (get_endpoints), indicating that the virtual environment is active.
3
Install Required Packages
Within the activated environment, install the necessary packages using pip. For this project, you need python-dotenv for managing environment variables and web3 for interacting with EVM blockchains:
Bash
pip install python-dotenv web3
It’s important to store your Chainstack API key securely. For this purpose, in your project’s root directory, create a .env file, which will hold sensitive configuration details like API keys, away from your main codebase.
Here’s how you can set it up:
1
Navigate to Your Project's Root Directory
Ensure you’re in the root directory of your project, where your main application files reside.
2
Create a `.env` File
In this directory, create a new file named .env. This file is commonly used to store environment variables.
3
Add Your Chainstack API Key
Open the .env file with a text editor and insert the following line:
env
CHAINSTACK_API_KEY="YOUR_API_KEY"
Replace YOUR_API_KEY with your actual Chainstack API key.
Now that you’ve set up your environment, the next step involves creating a Python script.
Follow these instructions:
1
Creating a New Python File in the Root Directory
Create a new Python file in your project’s root directory (the same location where you created the .env file). You can name this file get_endpoints.py.
This naming convention indicates the file’s purpose, which is to retrieve endpoint data.
2
Adding Code to the Python File
After creating get_endpoints.py, open it in your preferred code editor. You are now ready to add the necessary code. Paste the Python script.
This Python script is designed to interact with the Chainstack API, retrieve blockchain node information, and create a configuration file with these details.
- Environment Setup and Logging:
- The script starts by loading environment variables from a
.envfile, particularly theCHAINSTACK_API_KEY. This approach is a security best practice, keeping sensitive information from the source code. - Logging is initialized for recording the script’s activities, errors, and informational messages, which is crucial for monitoring and debugging.
- The script starts by loading environment variables from a
get_endpoints.py
import requests
import os
import logging
from dotenv import load_dotenv
from web3 import Web3
from typing import Optional, Dict, Any
# Load environment variables
load_dotenv()
# Constants
CHAINSTACK_API_KEY = os.getenv('CHAINSTACK_API_KEY')
OUTPUT_FILE_NAME = 'rpc.env'
# Initialize logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
- Fetching Data from Chainstack API:
- The
fetch_chainstack_datafunction makes a GET request to the Chainstack API using the provided API key. It retrieves information about the blockchain nodes on your account. - If successful, the function logs the number of items fetched and returns the data; otherwise, it logs an error and returns
None.
- The
get_endpoints.py
def fetch_chainstack_data(api_key: str) -> Optional[Dict[str, Any]]:
"""Fetch data from Chainstack API."""
url = "https://api.chainstack.com/v1/nodes/"
headers = {
"accept": "application/json",
"authorization": f"Bearer {api_key}"
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
logging.info(f"Fetched {len(data.get('results', []))} items from Chainstack.")
return data
except requests.RequestException as e:
logging.error(f"Failed to fetch data from Chainstack: {e}")
return None
- Processing Chainstack Data:
process_chainstack_itemtakes an item from the Chainstack data and extracts important details like ID, name, and endpoint information. This function is vital for organizing and simplifying the raw API data.
You can use this data as you wish.
get_endpoints.py
def process_chainstack_item(item: Dict[str, Any]) -> Dict[str, str]:
"""Process a single item from Chainstack data."""
logging.debug(f"Processing item: {item['name']} with ID {item['id']}")
return {
'id': item['id'],
'name': item['name'],
'details': item['details'],
'http_endpoint': item['details'].get('https_endpoint'),
'auth_key': item['details'].get('auth_key'),
'configuration': item['configuration'],
'client': item['configuration'].get('client')
}
- Web3 Connection Test:
- The
connect_to_web3function attempts to establish a connection to a Web3 endpoint. This step is crucial for verifying the functionality of the blockchain nodes as they are built from the HTTPS endpoint, and the AUTH key is fetched from the API. - It logs a successful connection or warns if it fails, providing valuable feedback about each endpoint’s status.
- The
Note that non-EVM endpoints or improperly formatted ones will not work and will show a warning. This is a point you can build on and improve.
get_endpoints.py
def connect_to_web3(reconstructed_endpoint: str) -> bool:
"""Connect to a Web3 endpoint."""
logging.debug(f"Attempting to connect to Web3 endpoint: {reconstructed_endpoint}")
try:
w3 = Web3(Web3.HTTPProvider(reconstructed_endpoint))
if w3.is_connected():
chain_id = w3.eth.chain_id
logging.info(f"Connected to {reconstructed_endpoint} with chain ID: {chain_id}")
return True
else:
logging.warning(f"Failed to connect to {reconstructed_endpoint}")
except Exception as e:
logging.error(f"An error occurred while connecting to {reconstructed_endpoint}: {e}")
return False
- Data Sanitization and .env File Creation:
sanitize_nameensures that the names of the endpoints are formatted correctly to be used as environment variable keys in the .env file.create_env_filewrites the endpoint URLs into a .env file, making them easily accessible and usable in other project parts. This file serves as a central configuration point, enhancing the modularity and scalability of the system.
get_endpoints.py
def sanitize_name(name: str) -> str:
"""Sanitize the endpoint name for use as an environment variable key."""
return name.replace(" ", "_").replace("-", "_").replace("/", "_").upper()
def create_env_file(endpoint_info_dict: Dict[str, Dict[str, str]], filename: str = OUTPUT_FILE_NAME) -> None:
"""Create a .env file from the endpoint info dictionary."""
logging.info(f"Preparing to write {len(endpoint_info_dict)} endpoints to .env file.")
with open(filename, 'w') as file:
for endpoint, info in endpoint_info_dict.items():
sanitized_name = sanitize_name(info['name'])
file.write(f'{sanitized_name}_URL="{endpoint}"\n')
logging.info(f".env file created successfully at {filename}.")
- Main Execution Flow:
- The
mainfunction orchestrates the entire process: checking the API key, fetching and processing data, testing endpoints, and creating the .env file. - It ensures each step is executed in sequence and handles the absence of data or API keys, making the script robust and user-friendly.
- The
get_endpoints.py
def main() -> None:
"""Main function to orchestrate the process."""
logging.info("Starting main process.")
if not CHAINSTACK_API_KEY:
logging.error("Chainstack API key not found.")
return
json_data = fetch_chainstack_data(CHAINSTACK_API_KEY)
if not json_data:
return
endpoint_info_dict = {}
for item in json_data.get('results', []):
data = process_chainstack_item(item)
reconstructed_endpoint = f"{data['http_endpoint']}/{data['auth_key']}"
if connect_to_web3(reconstructed_endpoint):
endpoint_info_dict[reconstructed_endpoint] = {'name': data['name']}
if endpoint_info_dict:
create_env_file(endpoint_info_dict)
else:
logging.info("No endpoint information to write to .env file.")
logging.info("Main process completed.")
if __name__ == "__main__":
main()
Executing this script will efficiently process and validate your Chainstack endpoints. It identifies those endpoints that successfully return a chain ID, indicating their active and functional status. These validated endpoints are then neatly recorded into a .env file.
With this setup, you gain a streamlined and organized method to access and utilize all your Chainstack endpoints.
This approach simplifies endpoint management and enhances the overall efficiency of your interactions with Chainstack services.
Note that non-EVM endpoints or improperly formatted ones will not work and will show a warning. This is a point you can build on and improve.
import requests
import os
import logging
from dotenv import load_dotenv
from web3 import Web3
from typing import Optional, Dict, Any
# Load environment variables
load_dotenv()
# Constants
CHAINSTACK_API_KEY = os.getenv('CHAINSTACK_API_KEY')
OUTPUT_FILE_NAME = 'rpc.env'
# Initialize logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def fetch_chainstack_data(api_key: str) -> Optional[Dict[str, Any]]:
"""Fetch data from Chainstack API."""
url = "https://api.chainstack.com/v1/nodes/"
headers = {
"accept": "application/json",
"authorization": f"Bearer {api_key}"
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
logging.info(f"Fetched {len(data.get('results', []))} items from Chainstack.")
return data
except requests.RequestException as e:
logging.error(f"Failed to fetch data from Chainstack: {e}")
return None
def process_chainstack_item(item: Dict[str, Any]) -> Dict[str, str]:
"""Process a single item from Chainstack data."""
logging.debug(f"Processing item: {item['name']} with ID {item['id']}")
return {
'id': item['id'],
'name': item['name'],
'details': item['details'],
'http_endpoint': item['details'].get('https_endpoint'),
'auth_key': item['details'].get('auth_key'),
'configuration': item['configuration'],
'client': item['configuration'].get('client')
}
def connect_to_web3(reconstructed_endpoint: str) -> bool:
"""Connect to a Web3 endpoint."""
logging.debug(f"Attempting to connect to Web3 endpoint: {reconstructed_endpoint}")
try:
w3 = Web3(Web3.HTTPProvider(reconstructed_endpoint))
if w3.is_connected():
chain_id = w3.eth.chain_id
logging.info(f"Connected to {reconstructed_endpoint} with chain ID: {chain_id}")
return True
else:
logging.warning(f"Failed to connect to {reconstructed_endpoint}")
except Exception as e:
logging.error(f"An error occurred while connecting to {reconstructed_endpoint}: {e}")
return False
def sanitize_name(name: str) -> str:
"""Sanitize the endpoint name for use as an environment variable key."""
return name.replace(" ", "_").replace("-", "_").replace("/", "_").upper()
def create_env_file(endpoint_info_dict: Dict[str, Dict[str, str]], filename: str = OUTPUT_FILE_NAME) -> None:
"""Create a .env file from the endpoint info dictionary."""
logging.info(f"Preparing to write {len(endpoint_info_dict)} endpoints to .env file.")
with open(filename, 'w') as file:
for endpoint, info in endpoint_info_dict.items():
sanitized_name = sanitize_name(info['name'])
file.write(f'{sanitized_name}_URL="{endpoint}"\n')
logging.info(f".env file created successfully at {filename}.")
def main() -> None:
"""Main function to orchestrate the process."""
logging.info("Starting main process.")
if not CHAINSTACK_API_KEY:
logging.error("Chainstack API key not found.")
return
json_data = fetch_chainstack_data(CHAINSTACK_API_KEY)
if not json_data:
return
endpoint_info_dict = {}
for item in json_data.get('results', []):
data = process_chainstack_item(item)
reconstructed_endpoint = f"{data['http_endpoint']}/{data['auth_key']}"
if connect_to_web3(reconstructed_endpoint):
endpoint_info_dict[reconstructed_endpoint] = {'name': data['name']}
if endpoint_info_dict:
create_env_file(endpoint_info_dict)
else:
logging.info("No endpoint information to write to .env file.")
logging.info("Main process completed.")
if __name__ == "__main__":
main()
2024-01-09 17:50:27,635 - INFO - Starting main process.
2024-01-09 17:50:29,556 - INFO - Fetched 9 items from Chainstack.
2024-01-09 17:50:30,707 - INFO - Connected to https://nd-422-757-666.p2pify.com/ with chain ID: 1
2024-01-09 17:50:33,324 - INFO - Connected to https://nd-828-700-214.p2pify.com/ with chain ID: 137
2024-01-09 17:50:34,953 - WARNING - Failed to connect to https://nd-418-459-126.p2pify.com/
2024-01-09 17:50:36,296 - INFO - Connected to https://nd-000-364-211.p2pify.com/ with chain ID: 42161
2024-01-09 17:50:38,214 - WARNING - Failed to connect to https://nd-326-444-187.p2pify.com/
2024-01-09 17:50:40,403 - INFO - Connected to https://nd-500-249-268.p2pify.com/ with chain ID: 100
2024-01-09 17:50:41,948 - INFO - Connected to https://nd-363-550-219.p2pify.com/ with chain ID: 1101
2024-01-09 17:50:42,927 - INFO - Connected to https://nd-954-882-037.p2pify.com/ with chain ID: 42161
2024-01-09 17:50:43,607 - WARNING - Failed to connect to https://starknet-mainnet.core.chainstack.com/
2024-01-09 17:50:43,607 - INFO - Preparing to write 6 endpoints to .env file.
2024-01-09 17:50:43,608 - INFO - .env file created successfully at rpc.env.
2024-01-09 17:50:43,608 - INFO - Main process completed.
Was this page helpful?
import requests
import os
import logging
from dotenv import load_dotenv
from web3 import Web3
from typing import Optional, Dict, Any
# Load environment variables
load_dotenv()
# Constants
CHAINSTACK_API_KEY = os.getenv('CHAINSTACK_API_KEY')
OUTPUT_FILE_NAME = 'rpc.env'
# Initialize logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def fetch_chainstack_data(api_key: str) -> Optional[Dict[str, Any]]:
"""Fetch data from Chainstack API."""
url = "https://api.chainstack.com/v1/nodes/"
headers = {
"accept": "application/json",
"authorization": f"Bearer {api_key}"
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
logging.info(f"Fetched {len(data.get('results', []))} items from Chainstack.")
return data
except requests.RequestException as e:
logging.error(f"Failed to fetch data from Chainstack: {e}")
return None
def process_chainstack_item(item: Dict[str, Any]) -> Dict[str, str]:
"""Process a single item from Chainstack data."""
logging.debug(f"Processing item: {item['name']} with ID {item['id']}")
return {
'id': item['id'],
'name': item['name'],
'details': item['details'],
'http_endpoint': item['details'].get('https_endpoint'),
'auth_key': item['details'].get('auth_key'),
'configuration': item['configuration'],
'client': item['configuration'].get('client')
}
def connect_to_web3(reconstructed_endpoint: str) -> bool:
"""Connect to a Web3 endpoint."""
logging.debug(f"Attempting to connect to Web3 endpoint: {reconstructed_endpoint}")
try:
w3 = Web3(Web3.HTTPProvider(reconstructed_endpoint))
if w3.is_connected():
chain_id = w3.eth.chain_id
logging.info(f"Connected to {reconstructed_endpoint} with chain ID: {chain_id}")
return True
else:
logging.warning(f"Failed to connect to {reconstructed_endpoint}")
except Exception as e:
logging.error(f"An error occurred while connecting to {reconstructed_endpoint}: {e}")
return False
def sanitize_name(name: str) -> str:
"""Sanitize the endpoint name for use as an environment variable key."""
return name.replace(" ", "_").replace("-", "_").replace("/", "_").upper()
def create_env_file(endpoint_info_dict: Dict[str, Dict[str, str]], filename: str = OUTPUT_FILE_NAME) -> None:
"""Create a .env file from the endpoint info dictionary."""
logging.info(f"Preparing to write {len(endpoint_info_dict)} endpoints to .env file.")
with open(filename, 'w') as file:
for endpoint, info in endpoint_info_dict.items():
sanitized_name = sanitize_name(info['name'])
file.write(f'{sanitized_name}_URL="{endpoint}"\n')
logging.info(f".env file created successfully at {filename}.")
def main() -> None:
"""Main function to orchestrate the process."""
logging.info("Starting main process.")
if not CHAINSTACK_API_KEY:
logging.error("Chainstack API key not found.")
return
json_data = fetch_chainstack_data(CHAINSTACK_API_KEY)
if not json_data:
return
endpoint_info_dict = {}
for item in json_data.get('results', []):
data = process_chainstack_item(item)
reconstructed_endpoint = f"{data['http_endpoint']}/{data['auth_key']}"
if connect_to_web3(reconstructed_endpoint):
endpoint_info_dict[reconstructed_endpoint] = {'name': data['name']}
if endpoint_info_dict:
create_env_file(endpoint_info_dict)
else:
logging.info("No endpoint information to write to .env file.")
logging.info("Main process completed.")
if __name__ == "__main__":
main()
2024-01-09 17:50:27,635 - INFO - Starting main process.
2024-01-09 17:50:29,556 - INFO - Fetched 9 items from Chainstack.
2024-01-09 17:50:30,707 - INFO - Connected to https://nd-422-757-666.p2pify.com/ with chain ID: 1
2024-01-09 17:50:33,324 - INFO - Connected to https://nd-828-700-214.p2pify.com/ with chain ID: 137
2024-01-09 17:50:34,953 - WARNING - Failed to connect to https://nd-418-459-126.p2pify.com/
2024-01-09 17:50:36,296 - INFO - Connected to https://nd-000-364-211.p2pify.com/ with chain ID: 42161
2024-01-09 17:50:38,214 - WARNING - Failed to connect to https://nd-326-444-187.p2pify.com/
2024-01-09 17:50:40,403 - INFO - Connected to https://nd-500-249-268.p2pify.com/ with chain ID: 100
2024-01-09 17:50:41,948 - INFO - Connected to https://nd-363-550-219.p2pify.com/ with chain ID: 1101
2024-01-09 17:50:42,927 - INFO - Connected to https://nd-954-882-037.p2pify.com/ with chain ID: 42161
2024-01-09 17:50:43,607 - WARNING - Failed to connect to https://starknet-mainnet.core.chainstack.com/
2024-01-09 17:50:43,607 - INFO - Preparing to write 6 endpoints to .env file.
2024-01-09 17:50:43,608 - INFO - .env file created successfully at rpc.env.
2024-01-09 17:50:43,608 - INFO - Main process completed.