PK!7yyairtable_schema/__init__.pyimport typing __version__ = "0.1.1" class AirtableCredentials(typing.NamedTuple): username: str password: str PK!F eeairtable_schema/cli.pyimport json import click from selenium import webdriver from selenium.webdriver.common.desired_capabilities import DesiredCapabilities from airtable_schema import AirtableCredentials from airtable_schema import fetch @click.group() @click.option( "--username", required=True, envvar="AIRTABLE_USER", help="Airtable username" ) @click.option("--password", envvar="AIRTABLE_PASSWORD", help="Airtable password") @click.option( "--stdin-password", is_flag=True, default=False, help="Read password in from stdin" ) @click.pass_context def main( ctx: click.core.Context, username: str, password: str, stdin_password: bool ) -> None: if not (password or stdin_password): raise click.UsageError("Must supply one of `password` or `stdin_password`") if stdin_password: password = input() ctx.obj = AirtableCredentials(username=username, password=password) @main.command() @click.option("--app-id", required=True, envvar="APP_ID", help="App ID for workspace") @click.option( "--remote-driver-address", envvar="REMOTE_DRIVER_HOST", help="Remote selenium host" ) @click.pass_context def schema(ctx: click.core.Context, app_id: str, remote_driver_address: str) -> None: if remote_driver_address: driver = webdriver.Remote( command_executor=remote_driver_address, desired_capabilities=DesiredCapabilities.FIREFOX, ) else: opts = webdriver.FirefoxOptions() opts.set_headless(True) driver = webdriver.Firefox(firefox_options=opts) schema = fetch.fetch_schema(ctx.obj, app_id, driver) driver.close() click.echo(schema) PK! 4airtable_schema/fetch.pyimport logging import selenium from selenium import webdriver from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.by import By from selenium.common.exceptions import TimeoutException from airtable_schema import AirtableCredentials WAIT_SECS_FOR_PAGE_LOAD = 15 # set higher/lower as needed for a slow webpage load # from Kai_Curry JS_SCRIPT_2 = """ return ( JSON.stringify( _.mapValues( application.tablesById, table => _.set( _.omit(table, ['sampleRows']), 'columns', _.map( table.columns, item => _.set( item, 'foreignTable', _.get(item, 'foreignTable.id') ) ) ) ) ) ) """ def fetch_schema( credentials: AirtableCredentials, app_id: str, driver: selenium.webdriver.remote.webdriver.WebDriver, ): driver.get("https://airtable.com/" + app_id + "/api/docs") user_field = driver.find_element_by_name("email") password_field = driver.find_element_by_name("password") user_field.send_keys(credentials.username) password_field.send_keys(credentials.password) password_field.submit() delay = WAIT_SECS_FOR_PAGE_LOAD # seconds try: WebDriverWait(driver, delay).until( EC.presence_of_element_located((By.CSS_SELECTOR, ".fields")) ) except TimeoutException: raise return driver.execute_script(JS_SCRIPT_2) PK!v nnairtable_schema/schema.pyimport json import typing class TypeOptions: def __init__(self, obj) -> None: self.obj = obj def __eq__(self, other) -> bool: return isinstance(other, TypeOptions) and self.obj == other.obj def __repr__(self) -> str: return f"TypeOptions(obj={repr(self.obj)})" def to_dict(self) -> typing.Dict: return {"obj": self.obj} @staticmethod def from_dict(options) -> "TypeOptions": return TypeOptions(options.get('obj')) @staticmethod def from_obj(obj) -> "TypeOptions": return TypeOptions(obj) class Aircolumn: def __init__(self, id, name, type, type_options) -> None: self.id = id self.name = name self.type = type self.type_options = type_options def __eq__(self, other) -> bool: return ( isinstance(other, Aircolumn) and self.id == other.id and self.name == other.name and self.type == other.type and self.type_options == other.type_options ) def __repr__(self) -> str: return f"Aircolumn(id={repr(self.id)}, name={repr(self.name)}, type={repr(self.type)}, type_options={repr(self.type_options)})" def to_dict(self) -> typing.Dict: return { "id": self.id, "name": self.name, "type": self.type, "type_options": self.type_options.to_dict(), } @staticmethod def from_dict(column: typing.Dict) -> "Aircolumn": type_options = TypeOptions.from_dict(column.get("type_options")) column["type_options"] = type_options return Aircolumn(**column) @staticmethod def from_schema(column: typing.Dict) -> "Aircolumn": return Aircolumn( id=column.get("id"), name=column.get("name"), type=column.get("type"), type_options=TypeOptions.from_obj(column.get("typeOptions")), ) class Airtable: def __init__(self, id, name, columns, primary_column_name) -> None: self.id = id self.name = name self.columns = columns self.primary_column_name = primary_column_name def __eq__(self, other) -> bool: return ( isinstance(other, Airtable) and self.id == other.id and self.name == other.name and self.columns == other.columns and self.primary_column_name == other.primary_column_name ) def __repr__(self) -> str: return f"Airtable(id={repr(self.id)}, name={repr(self.name)}, columns={repr(self.columns)}, primary_column_name={repr(self.primary_column_name)})" def to_dict(self) -> typing.Dict: return { "id": self.id, "name": self.name, "columns": [column.to_dict() for column in self.columns], "primary_column_name": self.primary_column_name, } @staticmethod def from_dict(schema: typing.Dict) -> "Airtable": columns = [Aircolumn.from_dict(column) for column in schema.get("columns")] schema["columns"] = columns return Airtable(**schema) @staticmethod def from_schema(schema: typing.Dict) -> "Airtable": return Airtable( id=schema.get("id"), name=schema.get("name"), columns=[Aircolumn.from_schema(column) for column in schema.get("columns")], primary_column_name=schema.get("primaryColumnName"), ) class AirtableSchema: def __init__(self, tables) -> None: self.tables = tables def __eq__(self, other) -> bool: return isinstance(other, AirtableSchema) and self.tables == other.tables def __repr__(self) -> str: return f"AirtableSchema(tables={repr(self.tables)})" def to_dict(self) -> typing.Dict: return {"tables": [table.to_dict() for table in self.tables]} @staticmethod def from_schema(schema_dict: typing.Dict) -> "AirtableSchema": return AirtableSchema( tables=[Airtable.from_schema(table) for table in schema_dict.values()] ) @staticmethod def from_dict(schema: typing.Dict): tables = [Airtable.from_dict(d) for d in schema.get("tables")] schema["tables"] = tables return AirtableSchema(**schema) @staticmethod def from_str(schema_str: str) -> "AirtableSchema": schema = json.loads(schema_str) return AirtableSchema.from_schema(schema) PK!Hϖ]/50airtable_schema-0.1.1.dist-info/entry_points.txtN+I/N.,()J,*ILI1Rz9Vy\\PK!H_zTT%airtable_schema-0.1.1.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]n0H*J>mlcAPK!Hp=A(airtable_schema-0.1.1.dist-info/METADATAn0D= ;T YJP$hM$4_V}ЋTxA?:i4G'oB!!KrT"=жRJؚ%Pp0 -QIW{X`M4'FKvAZȴ /Wi[v ;)7VddAwzdVAbZ],9a#~$kk2+:YmSsnoŹtCR !un jYY?F%V>MgF*e㗏BԜ5!PK!He&airtable_schema-0.1.1.dist-info/RECORD9sP>{}s#D;ڭMg$Qf_^|M~PK!7yyairtable_schema/__init__.pyPK!F eeairtable_schema/cli.pyPK! 4Kairtable_schema/fetch.pyPK!v nnairtable_schema/schema.pyPK!Hϖ]/50airtable_schema-0.1.1.dist-info/entry_points.txtPK!H_zTT%$ airtable_schema-0.1.1.dist-info/WHEELPK!Hp=A( airtable_schema-0.1.1.dist-info/METADATAPK!He&>"airtable_schema-0.1.1.dist-info/RECORDPKu$