aget_time() — langchain Function Reference

Architecture documentation for the aget_time() function in _sql_record_manager.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 1 called by 1

Entity Profile

LangChainCore→ ApiManagement→ aget_time() — langchain Function Reference

Dependency Diagram

graph TD
  3e4c4a44_922b_f6c5_2589_1953bed0156b["aget_time()"]
  9d7938c1_1c25_4cae_be80_9fc00a5ed077["SQLRecordManager"]
  3e4c4a44_922b_f6c5_2589_1953bed0156b -->|defined in| 9d7938c1_1c25_4cae_be80_9fc00a5ed077
  739947b3_1dd1_a014_7b81_04c8441e1e6c["aupdate()"]
  739947b3_1dd1_a014_7b81_04c8441e1e6c -->|calls| 3e4c4a44_922b_f6c5_2589_1953bed0156b
  d75b3a18_aeb2_cb98_8ca3_1eab11972c74["_amake_session()"]
  3e4c4a44_922b_f6c5_2589_1953bed0156b -->|calls| d75b3a18_aeb2_cb98_8ca3_1eab11972c74
  style 3e4c4a44_922b_f6c5_2589_1953bed0156b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/indexes/_sql_record_manager.py lines 220–251

    async def aget_time(self) -> float:
        """Get the current server time as a timestamp.

        Please note it's critical that time is obtained from the server since
        we want a monotonic clock.
        """
        async with self._amake_session() as session:
            # * SQLite specific implementation, can be changed based on dialect.
            # * For SQLite, unlike unixepoch it will work with older versions of SQLite.
            # ----
            # julianday('now'): Julian day number for the current date and time.
            # The Julian day is a continuous count of days, starting from a
            # reference date (Julian day number 0).
            # 2440587.5 - constant represents the Julian day number for January 1, 1970
            # 86400.0 - constant represents the number of seconds
            # in a day (24 hours * 60 minutes * 60 seconds)
            if self.dialect == "sqlite":
                query = text("SELECT (julianday('now') - 2440587.5) * 86400.0;")
            elif self.dialect == "postgresql":
                query = text("SELECT EXTRACT (EPOCH FROM CURRENT_TIMESTAMP);")
            else:
                msg = f"Not implemented for dialect {self.dialect}"
                raise NotImplementedError(msg)

            dt = (await session.execute(query)).scalar_one_or_none()

            if isinstance(dt, decimal.Decimal):
                dt = float(dt)
            if not isinstance(dt, float):
                msg = f"Unexpected type for datetime: {type(dt)}"
                raise AssertionError(msg)  # noqa: TRY004
            return dt