90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
from typing import NamedTuple
|
|
import tomllib
|
|
def dict_to_namedtuple(name, dictionary):
|
|
fields = {}
|
|
for key, value in dictionary.items():
|
|
if isinstance(value, dict):
|
|
fields[key] = dict_to_namedtuple(key.capitalize(), value)
|
|
else:
|
|
fields[key] = value
|
|
return NamedTuple(name, [(k, type(v)) for k, v in fields.items()])(**fields)
|
|
|
|
|
|
def load_config():
|
|
with open("creds.toml", "rb") as f:
|
|
config_dict = tomllib.load(f)
|
|
config = dict_to_namedtuple("Config", config_dict)
|
|
return config
|
|
import webbrowser
|
|
import requests
|
|
import threading
|
|
from flask import Flask, request
|
|
|
|
def authenticate(client_id, client_secret, scopes=None, port=5000):
|
|
"""
|
|
Opens the browser for OAuth authorization and returns token data.
|
|
|
|
Args:
|
|
client_id (str): Your Tastytrade client ID
|
|
client_secret (str): Your Tastytrade client secret
|
|
scopes (list[str]): Optional list of scopes (default: None)
|
|
port (int): Port for local callback server (default: 5000)
|
|
|
|
Returns:
|
|
dict: Token response JSON (access_token, refresh_token, etc.)
|
|
"""
|
|
redirect_uri = f"http://127.0.0.1:{port}/callback"
|
|
auth_url = "https://api.tastytrade.com/oauth/authorize"
|
|
token_url = "https://api.tastytrade.com/oauth/token"
|
|
|
|
app = Flask(__name__)
|
|
auth_code_container = {}
|
|
|
|
@app.route("/callback")
|
|
def callback():
|
|
code = request.args.get("code")
|
|
auth_code_container["code"] = code
|
|
return "Authorization successful. You may close this window."
|
|
|
|
# Start local server in background
|
|
def run_server():
|
|
app.run(port=port, debug=False, use_reloader=False)
|
|
threading.Thread(target=run_server, daemon=True).start()
|
|
|
|
# Build auth URL
|
|
params = {
|
|
"response_type": "code",
|
|
"client_id": client_id,
|
|
"redirect_uri": redirect_uri
|
|
}
|
|
if scopes:
|
|
params["scope"] = " ".join(scopes)
|
|
|
|
# Open browser
|
|
webbrowser.open(f"{auth_url}?{requests.compat.urlencode(params)}")
|
|
|
|
# Wait for code
|
|
print("Waiting for authorization...")
|
|
while "code" not in auth_code_container:
|
|
pass
|
|
|
|
# Exchange code for token
|
|
token_resp = requests.post(token_url, data={
|
|
"grant_type": "authorization_code",
|
|
"code": auth_code_container["code"],
|
|
"redirect_uri": redirect_uri,
|
|
"client_id": client_id,
|
|
"client_secret": client_secret
|
|
})
|
|
token_resp.raise_for_status()
|
|
|
|
return token_resp.json()
|
|
|
|
# Example usage
|
|
if __name__ == "__main__":
|
|
client_id = "YOUR_CLIENT_ID"
|
|
client_secret = "YOUR_CLIENT_SECRET"
|
|
token_data = authenticate(client_id, client_secret, scopes=["accounts", "orders"])
|
|
print(token_data)
|
|
|