utils.py 2.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. import datetime
  2. import decimal
  3. from .base import Database
  4. class InsertVar:
  5. """
  6. A late-binding cursor variable that can be passed to Cursor.execute
  7. as a parameter, in order to receive the id of the row created by an
  8. insert statement.
  9. """
  10. types = {
  11. "AutoField": int,
  12. "BigAutoField": int,
  13. "SmallAutoField": int,
  14. "IntegerField": int,
  15. "BigIntegerField": int,
  16. "SmallIntegerField": int,
  17. "PositiveBigIntegerField": int,
  18. "PositiveSmallIntegerField": int,
  19. "PositiveIntegerField": int,
  20. "BooleanField": int,
  21. "FloatField": Database.DB_TYPE_BINARY_DOUBLE,
  22. "DateTimeField": Database.DB_TYPE_TIMESTAMP,
  23. "DateField": Database.Date,
  24. "DecimalField": decimal.Decimal,
  25. }
  26. def __init__(self, field):
  27. internal_type = getattr(field, "target_field", field).get_internal_type()
  28. self.db_type = self.types.get(internal_type, str)
  29. self.bound_param = None
  30. def bind_parameter(self, cursor):
  31. self.bound_param = cursor.cursor.var(self.db_type)
  32. return self.bound_param
  33. def get_value(self):
  34. return self.bound_param.getvalue()
  35. class Oracle_datetime(datetime.datetime):
  36. """
  37. A datetime object, with an additional class attribute
  38. to tell oracledb to save the microseconds too.
  39. """
  40. input_size = Database.DB_TYPE_TIMESTAMP
  41. @classmethod
  42. def from_datetime(cls, dt):
  43. return Oracle_datetime(
  44. dt.year,
  45. dt.month,
  46. dt.day,
  47. dt.hour,
  48. dt.minute,
  49. dt.second,
  50. dt.microsecond,
  51. )
  52. class BulkInsertMapper:
  53. BLOB = "TO_BLOB(%s)"
  54. DATE = "TO_DATE(%s)"
  55. INTERVAL = "CAST(%s as INTERVAL DAY(9) TO SECOND(6))"
  56. NCLOB = "TO_NCLOB(%s)"
  57. NUMBER = "TO_NUMBER(%s)"
  58. TIMESTAMP = "TO_TIMESTAMP(%s)"
  59. types = {
  60. "AutoField": NUMBER,
  61. "BigAutoField": NUMBER,
  62. "BigIntegerField": NUMBER,
  63. "BinaryField": BLOB,
  64. "BooleanField": NUMBER,
  65. "DateField": DATE,
  66. "DateTimeField": TIMESTAMP,
  67. "DecimalField": NUMBER,
  68. "DurationField": INTERVAL,
  69. "FloatField": NUMBER,
  70. "IntegerField": NUMBER,
  71. "PositiveBigIntegerField": NUMBER,
  72. "PositiveIntegerField": NUMBER,
  73. "PositiveSmallIntegerField": NUMBER,
  74. "SmallAutoField": NUMBER,
  75. "SmallIntegerField": NUMBER,
  76. "TextField": NCLOB,
  77. "TimeField": TIMESTAMP,
  78. }
  79. def dsn(settings_dict):
  80. if settings_dict["PORT"]:
  81. host = settings_dict["HOST"].strip() or "localhost"
  82. return Database.makedsn(host, int(settings_dict["PORT"]), settings_dict["NAME"])
  83. return settings_dict["NAME"]