| 213 | |
| 214 | |
| 215 | class RunSqlRequest(IStep): |
| 216 | def __init__(self, name: Text): |
| 217 | self.__step = TStep(name=name) |
| 218 | self.__step.sql_request = TSqlRequest() |
| 219 | |
| 220 | def with_variables(self, **variables) -> "RunSqlRequest": |
| 221 | self.__step.variables.update(variables) |
| 222 | return self |
| 223 | |
| 224 | def with_db_config( |
| 225 | self, user=None, password=None, ip=None, port=None, database=None, psm=None |
| 226 | ): |
| 227 | if user: |
| 228 | self.__step.sql_request.db_config.user = user |
| 229 | if password: |
| 230 | self.__step.sql_request.db_config.password = password |
| 231 | if ip: |
| 232 | self.__step.sql_request.db_config.ip = ip |
| 233 | if port: |
| 234 | self.__step.sql_request.db_config.port = port |
| 235 | if database: |
| 236 | self.__step.sql_request.db_config.database = database |
| 237 | if psm: |
| 238 | self.__step.sql_request.db_config.psm = psm |
| 239 | return self |
| 240 | |
| 241 | def fetchone(self, sql) -> "RunSqlRequest": |
| 242 | self.__step.sql_request.method = SqlMethodEnum.FETCHONE |
| 243 | self.__step.sql_request.sql = sql |
| 244 | return self |
| 245 | |
| 246 | def fetchmany(self, sql, size) -> "RunSqlRequest": |
| 247 | self.__step.sql_request.method = SqlMethodEnum.FETCHMANY |
| 248 | self.__step.sql_request.sql = sql |
| 249 | self.__step.sql_request.size = size |
| 250 | return self |
| 251 | |
| 252 | def fetchall(self, sql) -> "RunSqlRequest": |
| 253 | self.__step.sql_request.method = SqlMethodEnum.FETCHALL |
| 254 | self.__step.sql_request.sql = sql |
| 255 | return self |
| 256 | |
| 257 | def update(self, sql) -> "RunSqlRequest": |
| 258 | self.__step.sql_request.method = SqlMethodEnum.UPDATE |
| 259 | self.__step.sql_request.sql = sql |
| 260 | return self |
| 261 | |
| 262 | def delete(self, sql) -> "RunSqlRequest": |
| 263 | self.__step.sql_request.method = SqlMethodEnum.DELETE |
| 264 | self.__step.sql_request.sql = sql |
| 265 | return self |
| 266 | |
| 267 | def insert(self, sql) -> "RunSqlRequest": |
| 268 | self.__step.sql_request.method = SqlMethodEnum.INSERT |
| 269 | self.__step.sql_request.sql = sql |
| 270 | return self |
| 271 | |
| 272 | def with_retry(self, retry_times, retry_interval) -> "RunSqlRequest": |