Create dynamic table model classes for the given schema type. Returns: For task schema: (table_cls,) For experience/sft/dpo schema: (meta_cls, blob_cls)
(table_name: str, schema_type: str)
| 134 | |
| 135 | |
| 136 | def _create_table_classes(table_name: str, schema_type: str): |
| 137 | """Create dynamic table model classes for the given schema type. |
| 138 | |
| 139 | Returns: |
| 140 | For task schema: (table_cls,) |
| 141 | For experience/sft/dpo schema: (meta_cls, blob_cls) |
| 142 | """ |
| 143 | from trinity.buffer.schema import SQL_SCHEMA |
| 144 | |
| 145 | if schema_type is None: |
| 146 | schema_type = "task" |
| 147 | |
| 148 | base_class = SQL_SCHEMA.get(schema_type) |
| 149 | |
| 150 | if schema_type == "task": |
| 151 | table_attrs = { |
| 152 | "__tablename__": table_name, |
| 153 | "__abstract__": False, |
| 154 | "__table_args__": {"keep_existing": True}, |
| 155 | } |
| 156 | table_cls = type(table_name, (base_class,), table_attrs) |
| 157 | return (table_cls,) |
| 158 | |
| 159 | meta_attrs = { |
| 160 | "__tablename__": table_name, |
| 161 | "__abstract__": False, |
| 162 | "__table_args__": {"keep_existing": True}, |
| 163 | } |
| 164 | meta_cls = type(f"{table_name}_meta", (base_class,), meta_attrs) |
| 165 | |
| 166 | blob_table_name = f"{table_name}_blob" |
| 167 | blob_attrs = { |
| 168 | "__tablename__": blob_table_name, |
| 169 | "__abstract__": False, |
| 170 | "__table_args__": {"keep_existing": True}, |
| 171 | } |
| 172 | blob_cls = type(f"{table_name}_blob", (BlobModel,), blob_attrs) |
| 173 | return (meta_cls, blob_cls) |
| 174 | |
| 175 | |
| 176 | async def init_async_engine(db_url: str, table_name: str, schema_type: Optional[str]) -> Tuple: |
no test coverage detected