refactor: Move serialize_datetime function
Move serialize_datetime function to be in it's own file Refactor #COG-170
This commit is contained in:
parent
2cedcbe670
commit
71c1374859
2 changed files with 20 additions and 24 deletions
|
|
@ -1,33 +1,17 @@
|
||||||
from typing import List, Optional, get_type_hints
|
|
||||||
from sqlalchemy import JSON, Column, Table, select, delete
|
|
||||||
|
|
||||||
from ..models.ScoredResult import ScoredResult
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
|
||||||
from ..vector_db_interface import VectorDBInterface, DataPoint
|
|
||||||
from sqlalchemy.orm import Mapped, mapped_column
|
|
||||||
from ..embeddings.EmbeddingEngine import EmbeddingEngine
|
|
||||||
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
|
||||||
from pgvector.sqlalchemy import Vector
|
from pgvector.sqlalchemy import Vector
|
||||||
|
from typing import List, Optional, get_type_hints
|
||||||
|
from sqlalchemy.orm import Mapped, mapped_column
|
||||||
|
from sqlalchemy import JSON, Column, Table, select, delete
|
||||||
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
||||||
|
|
||||||
|
from .serialize_datetime import serialize_datetime
|
||||||
|
from ..models.ScoredResult import ScoredResult
|
||||||
|
from ..vector_db_interface import VectorDBInterface, DataPoint
|
||||||
|
from ..embeddings.EmbeddingEngine import EmbeddingEngine
|
||||||
from ...relational.sqlalchemy.SqlAlchemyAdapter import SQLAlchemyAdapter
|
from ...relational.sqlalchemy.SqlAlchemyAdapter import SQLAlchemyAdapter
|
||||||
from ...relational.ModelBase import Base
|
from ...relational.ModelBase import Base
|
||||||
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
|
|
||||||
# TODO: Find better location for function
|
|
||||||
def serialize_datetime(data):
|
|
||||||
"""Recursively convert datetime objects in dictionaries/lists to ISO format."""
|
|
||||||
if isinstance(data, dict):
|
|
||||||
return {key: serialize_datetime(value) for key, value in data.items()}
|
|
||||||
elif isinstance(data, list):
|
|
||||||
return [serialize_datetime(item) for item in data]
|
|
||||||
elif isinstance(data, datetime):
|
|
||||||
return data.isoformat() # Convert datetime to ISO 8601 string
|
|
||||||
else:
|
|
||||||
return data
|
|
||||||
|
|
||||||
|
|
||||||
class PGVectorAdapter(SQLAlchemyAdapter, VectorDBInterface):
|
class PGVectorAdapter(SQLAlchemyAdapter, VectorDBInterface):
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,12 @@
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
def serialize_datetime(data):
|
||||||
|
"""Recursively convert datetime objects in dictionaries/lists to ISO format."""
|
||||||
|
if isinstance(data, dict):
|
||||||
|
return {key: serialize_datetime(value) for key, value in data.items()}
|
||||||
|
elif isinstance(data, list):
|
||||||
|
return [serialize_datetime(item) for item in data]
|
||||||
|
elif isinstance(data, datetime):
|
||||||
|
return data.isoformat() # Convert datetime to ISO 8601 string
|
||||||
|
else:
|
||||||
|
return data
|
||||||
Loading…
Add table
Reference in a new issue