scripts/string_cli.py
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""STRING Database CLI Wrapper.
Provides access to STRING API v12.0 endpoints.
"""
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "polite-http",
# ]
# ///
import argparse
import json
import os
import sys
import time
from typing import Any
import urllib.parse
from polite_http import http_client
CALLER_IDENTITY = 'google-science-skills'
URL_TEMPLATE = (
'https://version-{api_version}-0.string-db.org/api/{format_type}/{endpoint}'
)
_CLIENT = None
class _DummyResponse:
def __init__(self, content):
self.content = content
def json(self):
return json.loads(self.content.decode('utf-8'))
def call_api(url: str, params: dict[str, Any], output_file: str):
"""Calls the STRING API and writes the response to a file.
Args:
url: The API URL to call.
params: A dictionary of parameters to send with the request.
output_file: The file path where the API response will be written.
Returns:
A dummy response object with a .json() method if the call is successful.
Exits the program on API errors.
"""
params['caller_identity'] = CALLER_IDENTITY
data = urllib.parse.urlencode(params).encode('utf-8')
assert _CLIENT is not None
content = None
try:
content = _CLIENT.fetch_bytes(url, method='POST', data=data)
except http_client.HttpError as e:
print(f'API Error (HTTP {e.status_code}): {e}', file=sys.stderr)
sys.exit(1)
os.makedirs(os.path.dirname(os.path.abspath(output_file)), exist_ok=True)
with open(output_file, 'wb') as f:
f.write(content)
print(f'Success: Output written to {output_file}')
return _DummyResponse(content)
def main():
parser = argparse.ArgumentParser(description='STRING Database CLI wrapper')
parser.add_argument(
'--api_version', default='12', help='STRING API version (default: 12)'
)
subparsers = parser.add_subparsers(dest='command', required=True)
# Common parent parser for output
parent_out = argparse.ArgumentParser(add_help=False)
parent_out.add_argument(
'--output', required=True, help='File to write output to'
)
# Common parent for identifiers and species
parent_id = argparse.ArgumentParser(add_help=False)
parent_id.add_argument(
'--identifiers',
nargs='+',
required=True,
help='List of protein names/IDs',
)
parent_id.add_argument(
'--species', type=int, help='NCBI Taxon ID (e.g. 9606 for human)'
)
# map
p_map = subparsers.add_parser(
'map',
parents=[parent_out, parent_id],
help='Map identifiers to STRING IDs',
)
p_map.add_argument('--echo_query', type=int, choices=[0, 1], default=1)
# network
p_net = subparsers.add_parser(
'network', parents=[parent_out, parent_id], help='Get interaction network'
)
p_net.add_argument('--required_score', type=int, help='0-1000 threshold')
p_net.add_argument(
'--network_type', choices=['functional', 'physical'], default='functional'
)
p_net.add_argument('--add_nodes', type=int, help='Number of nodes to add')
# image
p_img = subparsers.add_parser(
'image', parents=[parent_out, parent_id], help='Get network image'
)
p_img.add_argument(
'--format', choices=['image', 'highres_image', 'svg'], default='image'
)
p_img.add_argument(
'--network_flavor',
choices=['evidence', 'confidence', 'actions'],
default='evidence',
)
p_img.add_argument('--add_color_nodes', type=int)
# partners
p_part = subparsers.add_parser(
'partners',
parents=[parent_out, parent_id],
help='Get interaction partners',
)
p_part.add_argument('--limit', type=int, help='Max partners per protein')
# homology
subparsers.add_parser(
'homology', parents=[parent_out, parent_id], help='Get homology scores'
)
# homology_best
p_homb = subparsers.add_parser(
'homology-best', parents=[parent_out, parent_id], help='Best homology hit'
)
p_homb.add_argument(
'--species_b', help='Comma-separated target species (e.g., 10090,7227)'
)
# enrichment
subparsers.add_parser(
'enrichment',
parents=[parent_out, parent_id],
help='Functional enrichment',
)
# functional_annotation
p_fa = subparsers.add_parser(
'functional-annotation',
parents=[parent_out, parent_id],
help='Functional annotations',
)
p_fa.add_argument('--allow_pubmed', type=int, choices=[0, 1], default=0)
# functional_terms (uses term_text instead of identifiers)
p_ft = subparsers.add_parser(
'functional-terms', parents=[parent_out], help='Search proteins by term'
)
p_ft.add_argument(
'--term_text', required=True, help='e.g. Melanoma, GO:0008543'
)
p_ft.add_argument('--species', type=int, default=9606)
# ppi_enrichment
subparsers.add_parser(
'ppi-enrichment', parents=[parent_out, parent_id], help='PPI enrichment'
)
# version
subparsers.add_parser(
'version', parents=[parent_out], help='Get STRING version'
)
# valuesranks API key
subparsers.add_parser(
'valuesranks-key',
parents=[parent_out],
help='Get API key for Values/Ranks',
)
# valuesranks submit
p_vr_sub = subparsers.add_parser(
'valuesranks-submit', parents=[parent_out], help='Submit Values/Ranks job'
)
p_vr_sub.add_argument('--api_key', required=True)
p_vr_sub.add_argument(
'--input_file', required=True, help='Tab-separated file of ID and value'
)
p_vr_sub.add_argument('--species', type=int, required=True)
p_vr_sub.add_argument('--ge_fdr', type=float, default=0.01)
# valuesranks status
p_vr_stat = subparsers.add_parser(
'valuesranks-status', parents=[parent_out], help='Check job status'
)
p_vr_stat.add_argument('--api_key', required=True)
p_vr_stat.add_argument('--job_id', help='Omit to list all jobs')
p_vr_stat.add_argument(
'--wait',
action='store_true',
help='Wait for job to complete and download final TSV',
)
args = parser.parse_args()
global _CLIENT
base_url = f'https://version-{args.api_version}-0.string-db.org/'
_CLIENT = http_client.HttpClient(base_url, qps=1)
def _url(endpoint, *, format_type='tsv'):
return URL_TEMPLATE.format(
api_version=args.api_version, format_type=format_type, endpoint=endpoint
)
requires_species = [
'map',
'network',
'image',
'partners',
'homology',
'homology-best',
'enrichment',
'functional-annotation',
'ppi-enrichment',
]
if args.command in requires_species and (
not hasattr(args, 'species') or args.species is None
):
print(
f"Error: Command '{args.command}' requires --species", file=sys.stderr
)
sys.exit(1)
params = {}
if hasattr(args, 'identifiers') and args.identifiers:
params['identifiers'] = '\r'.join(args.identifiers)
if hasattr(args, 'species') and args.species is not None:
params['species'] = args.species
if args.command == 'map':
params['echo_query'] = args.echo_query
call_api(_url('get_string_ids'), params, args.output)
elif args.command == 'network':
if args.required_score:
params['required_score'] = args.required_score
params['network_type'] = args.network_type
if args.add_nodes:
params['add_nodes'] = args.add_nodes
call_api(_url('network'), params, args.output)
elif args.command == 'image':
params['network_flavor'] = args.network_flavor
if args.add_color_nodes:
params['add_color_nodes'] = args.add_color_nodes
call_api(_url('network', format_type=args.format), params, args.output)
elif args.command == 'partners':
if args.limit:
params['limit'] = args.limit
call_api(_url('interaction_partners'), params, args.output)
elif args.command == 'homology-best':
if args.species_b:
params['species_b'] = args.species_b.replace(',', '\r')
call_api(_url('homology_best'), params, args.output)
elif args.command == 'functional-annotation':
params['allow_pubmed'] = args.allow_pubmed
call_api(_url('functional_annotation'), params, args.output)
elif args.command == 'functional-terms':
params['term_text'] = args.term_text
call_api(_url('functional_terms'), params, args.output)
elif args.command == 'valuesranks-key':
call_api(_url('get_api_key', format_type='json'), {}, args.output)
elif args.command in ['enrichment', 'homology', 'ppi-enrichment', 'version']:
call_api(_url(args.command.replace('-', '_')), params, args.output)
elif args.command == 'valuesranks-submit':
params['api_key'] = args.api_key
params['species'] = args.species
params['ge_fdr'] = args.ge_fdr
try:
with open(args.input_file, 'r') as f:
lines = f.readlines()
valid_lines = []
for i, line in enumerate(lines):
parts = line.strip().split('\t')
if len(parts) >= 2:
try:
float(parts[1])
valid_lines.append(f'{parts[0]}\t{parts[1]}')
except ValueError:
print(
f"Skipping line {i+1} as it doesn't contain a valid numeric"
f' value: {line.strip()}',
file=sys.stderr,
)
elif line.strip():
print(
f"Skipping line {i+1} as it doesn't contain at least 2"
f' tab-separated columns: {line.strip()}',
file=sys.stderr,
)
params['identifiers'] = '\r'.join(valid_lines)
if not params['identifiers']:
print('Error: No valid data found in input file.', file=sys.stderr)
sys.exit(1)
except OSError as e:
print(f'Error reading {args.input_file}: {e}', file=sys.stderr)
sys.exit(1)
url = _url('valuesranks_enrichment_submit', format_type='json')
call_api(url, params, args.output)
elif args.command == 'valuesranks-status':
url = _url('valuesranks_enrichment_status', format_type='json')
params['api_key'] = args.api_key
if args.job_id:
params['job_id'] = args.job_id
if not args.wait or not args.job_id:
call_api(url, params, args.output)
return
# Loop until the call completes.
while True:
response = call_api(url, params, args.output)
try:
data = response.json()
except ValueError:
print('Error parsing status response.', file=sys.stderr)
sys.exit(1)
if not isinstance(data, list) or len(data) == 0:
print(f'Unexpected response format: {data}', file=sys.stderr)
sys.exit(1)
job_info = data[0]
status = job_info.get('status')
if status == 'success':
if download_url := job_info.get('download_url'):
print(
f'Job success. Downloading from {download_url} to {args.output}'
)
try:
content = _CLIENT.fetch_bytes(download_url)
except http_client.HttpError as e:
print(
f'Download Error (HTTP {e.status_code}): {e}',
file=sys.stderr,
)
sys.exit(1)
with open(args.output, 'wb') as f:
f.write(content)
break
else:
print('Success but no download URL found.', file=sys.stderr)
break
elif status == 'running' or status == 'queued':
print(f'Job still {status}... waiting 5 seconds.')
time.sleep(5)
else:
print(f'Job failed or unknown status: {status}', file=sys.stderr)
sys.exit(1)
if __name__ == '__main__':
main()