[Python] Mypy error caused by Psycopg in Docker container

Discussão em 'Python' iniciado por Stack, Janeiro 20, 2025.

  1. Stack

    Stack Membro Participativo

    I am using mypy for linting in my python project within my container. I am using psycopg and SQLAlchemy. However, I'm encountering issues related to type checking of psycopg.

    When I run mypy src/datasets/database.py on my code, I get the following error related to psycopg:

    /usr/local/lib/python3.10/site-packages/psycopg/_cmodule.py:1: error: disable_error_code: Invalid error code(s): import-not-found [misc]
    Found 1 error in 1 file (checked 1 source file)


    I have mypy configured with the following setup in my pyproject.toml file:

    [[tool.mypy.overrides]]
    module = "psycopg.*"
    ignore_missing_imports = true


    My database.py is as follows

    from typing import Any, Dict

    import pandas as pd
    import sqlalchemy.orm.decl_api
    from kedro.extras.datasets.pandas.sql_dataset import SQLTableDataSet
    from kedro.utils import load_obj
    from sqlalchemy.dialects.postgresql import insert as pg_insert

    class SQLTableORMDataset(SQLTableDataSet):
    def __init__(
    self,
    orm_model: str,
    credentials: Dict[str, Any],
    load_args: Dict[str, Any] = None,
    save_args: Dict[str, Any] = None,
    ) -> None:
    self.orm_model: sqlalchemy.orm.decl_api.DeclarativeBase = load_obj(orm_model)

    super().__init__(table_name=self.table_name, credentials=credentials, load_args=load_args, save_args=save_args)

    @property
    def table_name(self):
    return self.orm_model.__table__.name

    @property
    def table_schema(self):
    return self.orm_model.__table__.schema

    def delete_rows(self):
    engine = self.engines[self._connection_str]
    with engine.begin() as conn:
    statement = sqlalchemy.delete(self.orm_model).where(sqlalchemy.text("true"))
    conn.execute(statement)

    def upsert_data(self, data: pd.DataFrame):
    engine = self.engines[self._connection_str]
    with engine.begin() as conn:
    for row in data.to_dict(orient="records"):
    stmt = pg_insert(self.orm_model).values(**row) # type: ignore

    on_conflict_stmt = stmt.on_conflict_do_update(
    index_elements=["foo"],
    set_=row,
    )
    conn.execute(on_conflict_stmt)

    Environment:


    • Python version: 3.10


    • psycopg2 version: 3.2.4


    • mypy version: 1.1.1


    • types-psycopg2 version: 2.9.21.20241019


    • sqlalchemy: ~2.0


    • psycopg[binary]: ~3.1
    Question:


    I’ve tried multiple troubleshooting steps, but I keep getting the same error related to psycopg2 and its type stubs. Does anyone know how to resolve this issue or if there’s a specific configuration required to make psycopg2 type stubs work with mypy in a Dockerised environment?

    Continue reading...

Compartilhe esta Página