Export tables schema to openapi

sqlite2rest/database.py

@@ -33,6 +33,10 @@ def get_record(self, table_name, key):
         col_names = [column[0] for column in self.cursor.description]
         return dict(zip(col_names, row))
 
+    def get_table_schema(self, table_name):
+        self.cursor.execute(f"PRAGMA table_info({table_name});")
+        columns = self.cursor.fetchall()
+        return {column[1]: column[2] for column in columns}  # Return a dictionary mapping column names to SQLite types
 
     def create_record(self, table_name, data):
         columns = ', '.join(data.keys())

sqlite2rest/openapi.py

@@ -37,16 +37,11 @@ def add_paging_parameters(operation_obj):
         }
     ]
 
-def add_operation_to_path(path_item, method, rule_str, primary_key_type):
+def add_operation_to_path(path_item, method, rule_str, primary_key_type, schema):
     operation = get_operation_summary(method)
     table_name = rule_str.split('/')[1]
     operation_obj = {
         "summary": f"{operation} the {table_name} table",
-        "responses": {
-            "200": {
-                "description": "OK"
-            }
-        }
     }
     if method == 'GET':
         if '<id>' in rule_str:
@@ -61,10 +56,70 @@ def add_operation_to_path(path_item, method, rule_str, primary_key_type):
                     }
                 }
             ]
+            add_response_to_operation(operation_obj, schema)
         else:
             add_paging_parameters(operation_obj)
+            add_response_to_operation(operation_obj, {"type": "array", "items": schema})
+    elif method == 'POST':
+        operation_obj["requestBody"] = {
+            "content": {
+                "application/json": {
+                    "schema": schema
+                }
+            }
+        }
+        add_response_to_operation(operation_obj, schema)
+    elif method == 'PUT':
+        operation_obj["parameters"] = [
+            {
+                "name": "id",
+                "in": "path",
+                "description": "The ID of the record to update",
+                "required": True,
+                "schema": {
+                    "type": primary_key_type,
+                }
+            }
+        ]
+        operation_obj["requestBody"] = {
+            "content": {
+                "application/json": {
+                    "schema": schema
+                }
+            }
+        }
+        add_response_to_operation(operation_obj, schema)
+    elif method == 'DELETE':
+        operation_obj["parameters"] = [
+            {
+                "name": "id",
+                "in": "path",
+                "description": "The ID of the record to delete",
+                "required": True,
+                "schema": {
+                    "type": primary_key_type,
+                }
+            }
+        ]
+        operation_obj["responses"] = {
+            "200": {
+                "description": "OK"
+            }
+        }
     path_item[method.lower()] = operation_obj
 
+def add_response_to_operation(operation_obj, schema):
+    operation_obj["responses"] = {
+        "200": {
+            "description": "OK",
+            "content": {
+                "application/json": {
+                    "schema": schema
+                }
+            }
+        }
+    }
+
 def sqlite_type_to_openapi_type(sqlite_type):
     """
     Convert SQLite data types to OpenAPI data types.
@@ -74,14 +129,16 @@ def sqlite_type_to_openapi_type(sqlite_type):
         return "integer"
     elif sqlite_type in ["REAL", "DOUBLE", "DOUBLE PRECISION", "FLOAT"]:
         return "number"
-    elif sqlite_type in ["TEXT", "CHARACTER", "VARCHAR", "VARYING CHARACTER", "NCHAR", "NATIVE CHARACTER", "NVARCHAR", "CLOB"]:
+    elif sqlite_type in ["TEXT", "CHARACTER", "VARCHAR", "VARYING CHARACTER", "NCHAR", "NATIVE CHARACTER", "CLOB"]:
+        return "string"
+    elif sqlite_type.startswith("NVARCHAR"):
         return "string"
     elif sqlite_type in ["BLOB"]:
-        return "string", "byte"
+        return "byte"
     elif sqlite_type in ["BOOLEAN"]:
         return "boolean"
     elif sqlite_type in ["DATE", "DATETIME"]:
-        return "string", "date-time"
+        return "string"
     else:
         return "string"
 
@@ -93,27 +150,38 @@ def generate_openapi_spec(db):
             "title": "SQLite2REST",
             "version": __version__
         },
-        "paths": {}
+        "paths": {},
+        "components": {
+            "schemas": {}
+        }
     }
 
-    # Generate paths for each route
-    for rule in current_app.url_map.iter_rules():
-        # Skip the static routes
-        if rule.endpoint in ('static', 'openapi'):
-            continue
+    # Generate schemas for each table
+    spec["components"] = {"schemas": {}}
+    for table_name in db.get_tables():
+        schema = db.get_table_schema(table_name)
+        spec["components"]["schemas"][table_name] = {
+            "type": "object",
+            "properties": {column: {"type": sqlite_type_to_openapi_type(sqlite_type)} for column, sqlite_type in schema.items()}
+        }
 
-        # Initialize the path item object
-        if str(rule) not in spec["paths"]:
-            spec["paths"][str(rule)] = {}
+    # Generate paths for each table
+    for table_name in db.get_tables():
+        # Get the primary key and schema for the table
+        _, primary_key_type = db.get_primary_key(table_name)
+        primary_key_type = sqlite_type_to_openapi_type(primary_key_type)
+        schema = spec["components"]["schemas"][table_name]
 
-        path_item = spec["paths"][str(rule)]
+        # Add the GET (all records), POST, PUT, and DELETE operations for the table
+        path_item = spec["paths"].setdefault(f'/{table_name}', {})
+        add_operation_to_path(path_item, 'GET', f'/{table_name}', primary_key_type, schema)
+        add_operation_to_path(path_item, 'POST', f'/{table_name}', primary_key_type, schema)
 
-        # Add an operation object for each method
-        for method in rule.methods:
-            if method in ['GET', 'POST', 'PUT', 'DELETE']:
-                table_name = str(rule).split('/')[1]
-                _, primary_key_type = db.get_primary_key(table_name)
-                add_operation_to_path(path_item, method, str(rule), sqlite_type_to_openapi_type(primary_key_type))
+        # Add the GET (single record), PUT, and DELETE operations for a record in the table
+        path_item = spec["paths"].setdefault(f'/{table_name}/<id>', {})
+        add_operation_to_path(path_item, 'GET', f'/{table_name}/<id>', primary_key_type, schema)
+        add_operation_to_path(path_item, 'PUT', f'/{table_name}/<id>', primary_key_type, schema)
+        add_operation_to_path(path_item, 'DELETE', f'/{table_name}/<id>', primary_key_type, schema)
 
     # Validate the spec
     validate_spec(spec)