def get_engine()

in IAC/temp/function-source/src/dbconector.py [0:0]

• 20 lines of code
• 1 McCabe index (conditional complexity)

    def get_engine(self):
        def getconn() -> pg8000.dbapi.Connection:
            # Conecta ao banco de dados Cloud SQL com autenticação IAM
            conn: pg8000.dbapi.Connection = self.connector.connect(
                self.instance_connection_name,
                "pg8000",
                user=self.iam_user,
                db=self.database,
                enable_iam_auth=True,  # Habilita autenticação IAM
                ip_type=self.ip_type
            )
            return conn

        # Cria o pool de conexões com SQLAlchemy
        engine = create_engine(
            "postgresql+pg8000://",
            creator=getconn,
            pool_size=5,
            max_overflow=2,
            pool_timeout=30,
            pool_recycle=1800  # Recicla a conexão a cada 30 minutos
        )
        return engine