# Upload processing
-@attrs.define(slots=False)
-class SQLCommand():
- '''
- An SQL command or commands
-
- Attributes:
- stmt The statement or statements, formatted for psycopg3 substitution
- args Tuple of arguments used to substitute when executed.
- ec(ex) Produces the exception to raise an instance of on failure
- Input:
- ex The exception raised by psycopg3
- '''
- stmt = attrs.field()
- args = attrs.field()
- ec = attrs.field(default=None)
-
- def _explain_encoding_error(self, ex):
- '''Return a SQLEncodingError instance
- '''
- if isinstance(ex, UnicodeEncodeError):
- return core_ex.SQLEncodingError(
- ex,
- ("Data cannot be represented in the database"
- " connection's client-side character encoding"),
- (f'The SQL statement is ({self.stmt}) and the data supplied'
- f' is ({self.tupl})'))
- if isinstance(ex, psycopg.errors.UntranslatableCharacter):
- return core_ex.SQLEncodingError(
- ex,
- ("Data cannot be represented in the"
- " character encoding of the database"),
- (f'The SQL statement is ({self.stmt}) and the data supplied'
- f' is ({self.tupl})'))
- return ex
-
- def execute(self, cur):
- '''
- Execute the sql statement.
-
- Input:
- cur A psycopg3 cursor
-
- Side effects:
- Does something in the db.
- Can raise a psycopg3 error
- '''
- try:
- cur.execute(self.stmt, self.args)
- except UnicodeEncodeError as ex:
- if self.ec is None:
- raise self._explain_encoding_error(ex)
- raise self.ec(self._explain_encoding_error(ex))
- except psycopg.errors.UntranslatableCharacter as ex:
- if self.ec is None:
- raise self._explain_encoding_error(ex)
- raise self.ec(self._explain_encoding_error(ex))
- except psycopg.errors.DatabaseError as ex:
- if self.ec is None:
- raise ex
- raise self.ec(ex)
-
-
-@attrs.define(slots=False)
-class LogSQLCommand(SQLCommand):
- '''An SQL command that logs success or failure.
-
- Attributes:
- stmt The statement, formatted for psycopg3 substitution
- args Tuple of arguments used to substitute when executed.
- ec(ex) Produces the exception to raise an instance of on failure
- Input:
- ex The exception raised by psycopg3
- log_success
- Logs success
- log_failure(ex)
- Logs failure
- Input:
- ex The exception to log
- '''
- log_success = attrs.field(default=None)
- log_failure = attrs.field(default=None)
-
- def execute(self, cur):
- '''
- Execute the sql statement.
-
- Input:
- cur A psycopg3 cursor
-
- Side effects:
- Does something in the db.
- Can raise a psycopg3 error
- '''
- try:
- super().execute(cur)
- except (core_ex.UploadError, psycopg.errors.DatabaseError) as ex:
- if self.log_failure:
- self.log_failure(ex)
- raise
- else:
- if self.log_success:
- self.log_success()
-
-
class UploadLine(object):
'''
Representation of a generic uploaded line
raise NotImplementedError()
-class SQLData(DBData):
- '''
- SQL statements returning no data that execute in the db.
-
- Attributes:
- stmts List of SQLCommand instances
- '''
- def __init__(self, stmts):
- '''
- stmts List of SQLCommand instances
- '''
- super().__init__()
- self.stmts = stmts
-
- def _thunk(self):
- for stmt in self.stmts:
- yield lambda: stmt
-
-
class UploadData(DBData):
'''Uploaded data file
pass
-class ExecuteSQL(DataLineProcessor):
- def __init__(self, ue, uh):
- '''
- ue UploadEngine instance
- uh UploadHandler instance
- cur psycopg3 cursor
- '''
- super().__init__(ue, uh)
-
- def eat(self, sqlc):
- '''
- Executes an sql command in the db.
-
- sqlc An SQLCommand instance (a command and it's args)
- '''
- sqlc.execute(self.cur)
-
-
@attrs.define(slots=False)
class DBHandler():
'''
--- /dev/null
+# Copyright (C) 2013, 2014, 2015, 2018, 2020, 2021, 2024 The Meme Factory, Inc.
+# http://www.karlpinc.com/
+
+# This file is part of PGWUI_Core.
+#
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public License
+# as published by the Free Software Foundation, either version 3 of
+# the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public
+# License along with this program. If not, see
+# <http://www.gnu.org/licenses/>.
+#
+
+# Karl O. Pinc <kop@karlpinc.com>
+
+'''SQL execution facilities.
+'''
+
+import attrs
+import psycopg.errors
+
+from . import core
+from . import exceptions as core_ex
+
+
+@attrs.define(slots=False)
+class SQLCommand():
+ '''
+ An SQL command or commands
+
+ Attributes:
+ stmt The statement or statements, formatted for psycopg3 substitution
+ args Tuple of arguments used to substitute when executed.
+ ec(ex) Produces the exception to raise an instance of on failure
+ Input:
+ ex The exception raised by psycopg3
+ '''
+ stmt = attrs.field()
+ args = attrs.field()
+ ec = attrs.field(default=None)
+
+ def _explain_encoding_error(self, ex):
+ '''Return a SQLEncodingError instance
+ '''
+ if isinstance(ex, UnicodeEncodeError):
+ return core_ex.SQLEncodingError(
+ ex,
+ ("Data cannot be represented in the database"
+ " connection's client-side character encoding"),
+ (f'The SQL statement is ({self.stmt}) and the data supplied'
+ f' is ({self.tupl})'))
+ if isinstance(ex, psycopg.errors.UntranslatableCharacter):
+ return core_ex.SQLEncodingError(
+ ex,
+ ("Data cannot be represented in the"
+ " character encoding of the database"),
+ (f'The SQL statement is ({self.stmt}) and the data supplied'
+ f' is ({self.tupl})'))
+ return ex
+
+ def execute(self, cur):
+ '''
+ Execute the sql statement.
+
+ Input:
+ cur A psycopg3 cursor
+
+ Side effects:
+ Does something in the db.
+ Can raise a psycopg3 error
+ '''
+ try:
+ cur.execute(self.stmt, self.args)
+ except UnicodeEncodeError as ex:
+ if self.ec is None:
+ raise self._explain_encoding_error(ex)
+ raise self.ec(self._explain_encoding_error(ex))
+ except psycopg.errors.UntranslatableCharacter as ex:
+ if self.ec is None:
+ raise self._explain_encoding_error(ex)
+ raise self.ec(self._explain_encoding_error(ex))
+ except psycopg.errors.DatabaseError as ex:
+ if self.ec is None:
+ raise ex
+ raise self.ec(ex)
+
+
+@attrs.define(slots=False)
+class LogSQLCommand(SQLCommand):
+ '''An SQL command that logs success or failure.
+
+ Attributes:
+ stmt The statement, formatted for psycopg3 substitution
+ args Tuple of arguments used to substitute when executed.
+ ec(ex) Produces the exception to raise an instance of on failure
+ Input:
+ ex The exception raised by psycopg3
+ log_success
+ Logs success
+ log_failure(ex)
+ Logs failure
+ Input:
+ ex The exception to log
+ '''
+ log_success = attrs.field(default=None)
+ log_failure = attrs.field(default=None)
+
+ def execute(self, cur):
+ '''
+ Execute the sql statement.
+
+ Input:
+ cur A psycopg3 cursor
+
+ Side effects:
+ Does something in the db.
+ Can raise a psycopg3 error
+ '''
+ try:
+ super().execute(cur)
+ except (core_ex.UploadError, psycopg.errors.DatabaseError) as ex:
+ if self.log_failure:
+ self.log_failure(ex)
+ raise
+ else:
+ if self.log_success:
+ self.log_success()
+
+
+class SQLData(core.DBData):
+ '''
+ SQL statements returning no data that execute in the db.
+
+ Attributes:
+ stmts List of SQLCommand instances
+ '''
+ def __init__(self, stmts):
+ '''
+ stmts List of SQLCommand instances
+ '''
+ super().__init__()
+ self.stmts = stmts
+
+ def _thunk(self):
+ for stmt in self.stmts:
+ yield lambda: stmt
+
+
+class ExecuteSQL(core.DataLineProcessor):
+ def __init__(self, ue, uh):
+ '''
+ ue UploadEngine instance
+ uh UploadHandler instance
+ cur psycopg3 cursor
+ '''
+ super().__init__(ue, uh)
+
+ def eat(self, sqlc):
+ '''
+ Executes an sql command in the db.
+
+ sqlc An SQLCommand instance (a command and it's args)
+ '''
+ sqlc.execute(self.cur)
projects / pgwui_core / commitdiff