jamesp/oracle-livemount-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Consolidate RSC funcs

Browse Source
This commit is contained in:
James Pattinson
2025-11-14 17:45:11 +00:00
parent 7c0dfd0e14
commit 33358c76ec
6 changed files with 111 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
clone_oracle_database.py
View File

@@ -6,8 +6,7 @@ import os
import argparse import argparse
import time import time
from datetime import datetime from datetime import datetime
from rsc_auth import RSCAuth from rsc import RSCAuth, RSCGraphQL
from rsc_graphql import RSCGraphQL
def parse_options_file(options_file): def parse_options_file(options_file):
"""Parse the options file containing Oracle parameters""" """Parse the options file containing Oracle parameters"""

3
introspect_schema.py
View File

@@ -2,8 +2,7 @@
import json import json
import sys import sys
from rsc_auth import RSCAuth from rsc import RSCAuth, RSCGraphQL
from rsc_graphql import RSCGraphQL
def explore_oracle_types(): def explore_oracle_types():
"""Explore Oracle-related types in the GraphQL schema""" """Explore Oracle-related types in the GraphQL schema"""

3
list_db_snapshots.py
View File

@@ -4,8 +4,7 @@ import json
import sys import sys
import os import os
from datetime import datetime from datetime import datetime
from rsc_auth import RSCAuth from rsc import RSCAuth, RSCGraphQL
from rsc_graphql import RSCGraphQL
from tabulate import tabulate from tabulate import tabulate
def find_database_by_name_or_id(identifier): def find_database_by_name_or_id(identifier):

3
list_oracle_databases.py
View File

@@ -3,8 +3,7 @@
import json import json
import sys import sys
import os import os
from rsc_auth import RSCAuth from rsc import RSCAuth, RSCGraphQL
from rsc_graphql import RSCGraphQL
from tabulate import tabulate from tabulate import tabulate
def list_oracle_databases(): def list_oracle_databases():

106
rsc_auth.py → rsc.py
View File

@@ -79,3 +79,109 @@ class RSCAuth:
'Authorization': f'Bearer {self.get_token()}', 'Authorization': f'Bearer {self.get_token()}',
'Content-Type': 'application/json' 'Content-Type': 'application/json'
} }
class RSCGraphQL:
def __init__(self, auth):
self.auth = auth
self.endpoint = f"https://{self.auth.host}/api/graphql"
def query(self, query, variables=None):
payload = {'query': query}
if variables:
payload['variables'] = variables
headers = self.auth.get_headers()
response = requests.post(self.endpoint, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
# Check for GraphQL errors
if 'errors' in data:
raise Exception(f"GraphQL errors: {data['errors']}")
return data
def introspect_schema(self):
"""Introspect the GraphQL schema to get type information"""
introspection_query = """
query IntrospectionQuery {
__schema {
types {
name
kind
description
fields(includeDeprecated: true) {
name
description
type {
name
kind
ofType {
name
kind
}
}
args {
name
description
type {
name
kind
ofType {
name
kind
}
}
}
}
}
}
}
"""
return self.query(introspection_query)
def get_type_info(self, type_name):
"""Get detailed information about a specific GraphQL type"""
query = """
query GetTypeInfo($typeName: String!) {
__type(name: $typeName) {
name
kind
description
fields(includeDeprecated: true) {
name
description
type {
name
kind
ofType {
name
kind
ofType {
name
kind
}
}
}
args {
name
description
type {
name
kind
ofType {
name
kind
ofType {
name
kind
}
}
}
}
}
}
}
"""
return self.query(query, {"typeName": type_name})

108
rsc_graphql.py
View File

@@ -1,108 +0,0 @@
import json
import requests
class RSCGraphQL:
def __init__(self, auth):
self.auth = auth
self.endpoint = f"https://{self.auth.host}/api/graphql"
def query(self, query, variables=None):
payload = {'query': query}
if variables:
payload['variables'] = variables
headers = self.auth.get_headers()
response = requests.post(self.endpoint, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
# Check for GraphQL errors
if 'errors' in data:
raise Exception(f"GraphQL errors: {data['errors']}")
return data
def introspect_schema(self):
"""Introspect the GraphQL schema to get type information"""
introspection_query = """
query IntrospectionQuery {
__schema {
types {
name
kind
description
fields(includeDeprecated: true) {
name
description
type {
name
kind
ofType {
name
kind
}
}
args {
name
description
type {
name
kind
ofType {
name
kind
}
}
}
}
}
}
}
"""
return self.query(introspection_query)
def get_type_info(self, type_name):
"""Get detailed information about a specific GraphQL type"""
query = """
query GetTypeInfo($typeName: String!) {
__type(name: $typeName) {
name
kind
description
fields(includeDeprecated: true) {
name
description
type {
name
kind
ofType {
name
kind
ofType {
name
kind
}
}
}
args {
name
description
type {
name
kind
ofType {
name
kind
ofType {
name
kind
}
}
}
}
}
}
}
"""
return self.query(query, {"typeName": type_name})