Coverage for mindsdb / integrations / handlers / monetdb_handler / utils / monet_get_id.py: 0%

1from sqlalchemy import exc 

2 

3 

4def schema_id(connection, schema_name=None): 

5 """Fetch the id for schema""" 

6 cur = connection.cursor() 

7 if schema_name is None: 

8 cur.execute("SELECT current_schema") 

9 schema_name = cur.fetchall()[0][0] 

10 

11 query = f""" 

12 SELECT id 

13 FROM sys.schemas 

14 WHERE name = '{schema_name}' 

15 """ 

16 

17 cur.execute(query) 

18 

19 try: 

20 schema_id = cur.fetchall()[0][0] 

21 except Exception: 

22 raise exc.InvalidRequestError(schema_name) 

23 

24 return schema_id 

25 

26 

27def table_id(connection, table_name, schema_name=None): 

28 """Fetch the id for schema.table_name, defaulting to current schema if 

29 schema is None 

30 """ 

31 

32 schema_idm = schema_id(connection=connection, schema_name=schema_name) 

33 

34 q = f""" 

35 SELECT id 

36 FROM sys.tables 

37 WHERE name = '{table_name}' 

38 AND schema_id = {schema_idm} 

39 """ 

40 

41 cur = connection.cursor() 

42 cur.execute(q) 

43 

44 try: 

45 table_id = cur.fetchall()[0][0] 

46 except Exception: 

47 raise exc.NoSuchTableError(table_name) 

48 

49 return table_id