generated from MasoniteFramework/starter-package
-
-
Notifications
You must be signed in to change notification settings - Fork 48
/
Model.py
796 lines (616 loc) · 22.3 KB
/
Model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
import json
from datetime import datetime
from inflection import tableize
import inspect
from ..query import QueryBuilder
from ..collection import Collection
from ..observers import ObservesEvents
from ..scopes import TimeStampsMixin
"""This is a magic class that will help using models like User.first() instead of having to instatiate a class like
User().first()
"""
class ModelMeta(type):
def __getattr__(self, attribute, *args, **kwargs):
"""This method is called between a Model and accessing a property. This is a quick and easy
way to instantiate a class before the first method is called. This is to avoid needing
to do this:
User().where(..)
and instead, with this class inherited as a meta class, we can do this:
User.where(...)
This class (potentially magically) instantiates the class even though we really didn't instantiate it.
Args:
attribute (string): The name of the attribute
Returns:
Model|mixed: An instantiated model's attribute
"""
instantiated = self()
return getattr(instantiated, attribute)
class BoolCast:
"""Casts a value to a boolean"""
def get(self, value):
return bool(value)
def set(self, value):
return bool(value)
class JsonCast:
"""Casts a value to JSON"""
def get(self, value):
if isinstance(value, dict):
return value
return json.loads(value)
def set(self, value):
return json.dumps(value)
class IntCast:
"""Casts a value to a int"""
def get(self, value):
return int(value)
def set(self, value):
return int(value)
class FloatCast:
"""Casts a value to a float"""
def get(self, value):
return float(value)
def set(self, value):
return float(value)
class Model(TimeStampsMixin, ObservesEvents, metaclass=ModelMeta):
"""The ORM Model class
Base Classes:
TimeStampsMixin (TimeStampsMixin): Adds scopes to add timestamps when something is inserted
metaclass (ModelMeta, optional): Helps instantiate a class when it hasn't been instantiated. Defaults to ModelMeta.
"""
__fillable__ = ["*"]
__guarded__ = ["*"]
__dry__ = False
__table__ = None
__connection__ = "default"
__resolved_connection__ = None
__selects__ = []
__observers__ = []
_booted = False
_scopes = {}
__primary_key__ = "id"
__casts__ = {}
__dates__ = []
__hidden__ = []
__visible__ = []
__timestamps__ = True
__timezone__ = "UTC"
__with__ = ()
__force_update__ = False
date_created_at = "created_at"
date_updated_at = "updated_at"
"""Pass through will pass any method calls to the model directly through to the query builder.
Anytime one of these methods are called on the model it will actually be called on the query builder class.
"""
__passthrough__ = [
"add_select",
"all",
"avg",
"between",
"bulk_create",
"chunk",
"count",
"delete",
"find_or_404",
"find_or_fail",
"first_or_fail",
"first",
"force_update",
"get",
"has",
"join_on",
"join",
"joins",
"last",
"limit",
"max",
"min",
"not_between",
"or_where",
"order_by",
"paginate",
"select",
"set_global_scope",
"simple_paginate",
"statement",
"sum",
"to_qmark",
"to_sql",
"truncate",
"update",
"when",
"where_between",
"where_from_builder",
"where_has",
"where_in",
"where_like",
"where_not_between",
"where_not_like",
"where_not_null",
"where_null",
"where_raw",
"where",
"with_",
]
__cast_map__ = {}
__internal_cast_map__ = {
"bool": BoolCast,
"json": JsonCast,
"int": IntCast,
"float": FloatCast,
}
def __init__(self):
self.__attributes__ = {}
self.__original_attributes__ = {}
self.__dirty_attributes__ = {}
if not hasattr(self, "__appends__"):
self.__appends__ = []
self._relationships = {}
self._global_scopes = {}
self.get_builder()
self.boot()
@classmethod
def get_primary_key(self):
"""Gets the primary key column
Returns:
mixed
"""
return self.__primary_key__
def get_primary_key_value(self):
"""Gets the primary key value.
Raises:
AttributeError: Raises attribute error if the model does not have an
attribute with the primary key.
Returns:
str|int
"""
try:
return getattr(self, self.get_primary_key())
except AttributeError:
name = self.__class__.__name__
raise AttributeError(
f"class '{name}' has no attribute {self.get_primary_key()}. Did you set the primary key correctly on the model using the __primary_key__ attribute?"
)
def query(self):
return self.get_builder()
def get_builder(self):
if hasattr(self, "builder"):
return self.builder
self.builder = QueryBuilder(
connection=self.__connection__,
table=self.get_table_name(),
connection_details=self.get_connection_details(),
model=self,
scopes=self._scopes,
dry=self.__dry__,
)
return self.builder.select(*self.__selects__)
def get_connection_details(self):
from config.database import ConnectionResolver
return ConnectionResolver().get_connection_details()
def boot(self):
if not self._booted:
self.observe_events(self, "booting")
for base_class in inspect.getmro(self.__class__):
class_name = base_class.__name__
if class_name.endswith("Mixin"):
getattr(base_class(), "boot_" + class_name)(self.builder)
self._booted = True
self.observe_events(self, "booted")
self.append_passthrough(list(self.builder._macros.keys()))
def append_passthrough(self, passthrough):
self.__passthrough__ += passthrough
return self
@classmethod
def get_table_name(cls):
"""Gets the table name.
Returns:
str
"""
return cls.__table__ or tableize(cls.__name__)
@classmethod
def table(cls, table):
"""Gets the table name.
Returns:
str
"""
cls.__table__ = table
return cls
@classmethod
def find(cls, record_id, query=False):
"""Finds a row by the primary key ID.
Arguments:
record_id {int} -- The ID of the primary key to fetch.
Returns:
Model
"""
if isinstance(record_id, (list, tuple)):
builder = cls().where_in(cls.get_primary_key(), record_id)
else:
builder = cls().where(cls.get_primary_key(), record_id)
if query:
return builder.to_sql()
else:
return builder.first()
def first_or_new(self):
pass
def first_or_create(self):
pass
def is_loaded(self):
return bool(self.__attributes__)
def is_created(self):
return self.get_primary_key() in self.__attributes__
def add_relation(self, relations):
self._relationships.update(relations)
return self
@classmethod
def hydrate(cls, result, relations=None):
"""Takes a result and loads it into a model
Args:
result ([type]): [description]
relations (dict, optional): [description]. Defaults to {}.
Returns:
[type]: [description]
"""
relations = relations or {}
if result is None:
return None
if isinstance(result, (list, tuple)):
response = []
for element in result:
response.append(cls.hydrate(element))
return cls.new_collection(response)
elif isinstance(result, dict):
model = cls()
dic = {}
for key, value in result.items():
if key in model.get_dates() and value:
value = model.get_new_date(value)
dic.update({key: value})
model.observe_events(model, "hydrating")
model.__attributes__.update(dic or {})
model.__original_attributes__.update(dic or {})
model.add_relation(relations)
model.observe_events(model, "hydrated")
return model
elif hasattr(result, "serialize"):
model = cls()
model.__attributes__.update(result.serialize())
model.__original_attributes__.update(result.serialize())
return model
else:
model = cls()
model.observe_events(model, "hydrating")
model.__attributes__.update(dict(result))
model.__original_attributes__.update(dict(result))
model.observe_events(model, "hydrated")
return model
def fill(self, attributes):
self.__attributes__.update(attributes)
self.__original_attributes__.update(attributes)
return self
@classmethod
def new_collection(cls, data):
"""Takes a result and puts it into a new collection.
This is designed to be able to be overidden by the user.
Args:
data (list|dict): Could be any data type but will be loaded directly into a collection.
Returns:
Collection
"""
return Collection(data)
@classmethod
def create(cls, dictionary=None, query=False, **kwargs):
"""Creates new records based off of a dictionary as well as data set on the model
such as fillable values.
Args:
dictionary (dict, optional): [description]. Defaults to {}.
query (bool, optional): [description]. Defaults to False.
Returns:
self: A hydrated version of a model
"""
if not dictionary:
dictionary = kwargs
if cls.__fillable__ != ["*"]:
d = {}
for x in cls.__fillable__:
if x in dictionary:
d.update({x: dictionary[x]})
dictionary = d
if cls.__guarded__ != ["*"]:
for x in cls.__guarded__:
if x in dictionary:
dictionary.pop(x)
if query:
return cls.builder.create(dictionary, query=True).to_sql()
return cls.builder.create(dictionary)
def fresh(self):
return (
self.get_builder()
.where(self.get_primary_key(), self.get_primary_key_value())
.first()
)
def serialize(self):
"""Takes the data as a model and converts it into a dictionary.
Returns:
dict
"""
serialized_dictionary = self.__attributes__
# prevent using both hidden and visible at the same time
if self.__visible__ and self.__hidden__:
raise AttributeError(
f"class model '{self.__class__.__name__}' defines both __visible__ and __hidden__."
)
if self.__visible__:
new_serialized_dictionary = {
k: serialized_dictionary[k]
for k in self.__visible__
if k in serialized_dictionary
}
serialized_dictionary = new_serialized_dictionary
else:
for key in self.__hidden__:
if key in serialized_dictionary:
serialized_dictionary.pop(key)
for date_column in self.get_dates():
if (
date_column in serialized_dictionary
and serialized_dictionary[date_column]
):
serialized_dictionary[date_column] = self.get_new_serialized_date(
serialized_dictionary[date_column]
)
serialized_dictionary.update(self.__dirty_attributes__)
# The builder is inside the attributes but should not be serialized
if "builder" in serialized_dictionary:
serialized_dictionary.pop("builder")
# Serialize relationships as well
serialized_dictionary.update(self.relations_to_dict())
for append in self.__appends__:
serialized_dictionary.update({append: getattr(self, append)})
for key, value in serialized_dictionary.items():
if isinstance(value, datetime):
value = self.get_new_serialized_date(value)
if key in self.__casts__:
value = self._cast_attribute(key, value)
serialized_dictionary.update({key: value})
return serialized_dictionary
def to_json(self):
"""Converts a model to JSON
Returns:
string
"""
return json.dumps(self.serialize())
@classmethod
def update_or_create(cls, wheres, updates):
self = cls()
record = self.where(wheres).first()
total = {}
total.update(updates)
total.update(wheres)
if not record:
return self.create(total)
return self.where(wheres).update(total)
def relations_to_dict(self):
"""Converts a models relationships to a dictionary
Returns:
[type]: [description]
"""
new_dic = {}
for key, value in self._relationships.items():
if value == {}:
new_dic.update({key: {}})
else:
if value is None:
new_dic.update({key: {}})
continue
new_dic.update({key: value.serialize()})
return new_dic
def touch(self, date=None, query=True):
"""Updates the current timestamps on the model"""
if not self.__timestamps__:
return False
self._update_timestamps(date=date)
return self.save(query=query)
def _update_timestamps(self, date=None):
"""Sets the updated at date to the current time or a specified date
Args:
date (datetime.datetime, optional): a date. If none is specified then it will use the current date Defaults to None.
"""
self.updated_at = date or self._current_timestamp()
def _current_timestamp(self):
return datetime.now()
def __getattr__(self, attribute):
"""Magic method that is called when an attribute does not exist on the model.
Args:
attribute (string): the name of the attribute being accessed or called.
Returns:
mixed: Could be anything that a method can return.
"""
if attribute in self.__passthrough__:
def method(*args, **kwargs):
return getattr(self.get_builder(), attribute)(*args, **kwargs)
return method
new_name_accessor = "get_" + attribute + "_attribute"
if (new_name_accessor) in self.__class__.__dict__:
return self.__class__.__dict__.get(new_name_accessor)(self)
if (
"__dirty_attributes__" in self.__dict__
and attribute in self.__dict__["__dirty_attributes__"]
):
return self.get_dirty_value(attribute)
if (
"__attributes__" in self.__dict__
and attribute in self.__dict__["__attributes__"]
):
if attribute in self.get_dates():
return (
self.get_new_date(self.get_value(attribute))
if self.get_value(attribute)
else None
)
return self.get_value(attribute)
if attribute in self.__dict__.get("_relationships", {}):
return self.__dict__["_relationships"][attribute]
if attribute not in self.__dict__:
name = self.__class__.__name__
raise AttributeError(f"class model '{name}' has no attribute {attribute}")
return None
def __setattr__(self, attribute, value):
if hasattr(self, "set_" + attribute + "_attribute"):
method = getattr(self, "set_" + attribute + "_attribute")
value = method(value)
if attribute in self.__casts__:
value = self._set_cast_attribute(attribute, value)
try:
if not attribute.startswith("_"):
self.__dict__["__dirty_attributes__"].update({attribute: value})
else:
self.__dict__[attribute] = value
except KeyError:
pass
def get_raw_attribute(self, attribute):
"""Gets an attribute without having to call the models magic methods. Gets around infinite recursion loops.
Args:
attribute (string): The attribute to fetch
Returns:
mixed: Any value an attribute can be.
"""
return self.__attributes__.get(attribute)
def is_dirty(self):
return bool(self.__dirty_attributes__)
def get_original(self, key):
return self.__original_attributes__.get(key)
def get_dirty(self, key):
return self.__dirty_attributes__.get(key)
def get_dirty_keys(self):
return list(self.get_dirty_attributes().keys())
def save(self, query=False):
builder = self.get_builder()
if "builder" in self.__dirty_attributes__:
self.__dirty_attributes__.pop("builder")
self.observe_events(self, "saving")
if not query:
if self.is_loaded():
result = builder.update(self.__dirty_attributes__)
else:
result = self.create(self.__dirty_attributes__, query=query)
self.observe_events(self, "saved")
self.fill(result.__attributes__)
return result
if self.is_loaded():
result = builder.update(self.__dirty_attributes__, dry=query).to_sql()
else:
result = self.create(self.__dirty_attributes__, query=query)
return result
def get_value(self, attribute):
value = self.__attributes__[attribute]
if attribute in self.__casts__:
return self._cast_attribute(attribute, value)
return value
def get_dirty_value(self, attribute):
value = self.__dirty_attributes__[attribute]
if attribute in self.__casts__:
return self._cast_attribute(attribute, value)
return value
def all_attributes(self):
attributes = self.__attributes__
attributes.update(self.get_dirty_attributes())
for key, value in attributes.items():
if key in self.__casts__:
attributes.update({key: self._cast_attribute(key, value)})
return attributes
def get_dirty_attributes(self):
if "builder" in self.__dirty_attributes__:
self.__dirty_attributes__.pop("builder")
return self.__dirty_attributes__ or {}
def get_cast_map(self):
cast_map = self.__internal_cast_map__
cast_map.update(self.__cast_map__)
return cast_map
def _cast_attribute(self, attribute, value):
cast_method = self.__casts__[attribute]
cast_map = self.get_cast_map()
if isinstance(cast_method, str):
return cast_map[cast_method]().get(value)
return cast_method(value)
def _set_cast_attribute(self, attribute, value):
cast_method = self.__casts__[attribute]
cast_map = self.get_cast_map()
if isinstance(cast_method, str):
return cast_map[cast_method]().set(value)
return cast_method(value)
@classmethod
def load(cls, *loads):
cls.boot()
cls._loads += loads
return cls.builder
def __getitem__(self, attribute):
return getattr(self, attribute)
def get_dates(self):
"""
Get the attributes that should be converted to dates.
:rtype: list
"""
defaults = [self.date_created_at, self.date_updated_at]
return self.__dates__ + defaults
def get_new_date(self, datetime=None):
"""
Get the attributes that should be converted to dates.
:rtype: list
"""
import pendulum
if not datetime:
return pendulum.now(self.__timezone__)
if isinstance(datetime, str):
return pendulum.parse(datetime, tz=self.__timezone__)
return pendulum.instance(datetime, tz=self.__timezone__)
def get_new_datetime_string(self, datetime=None):
"""
Get the attributes that should be converted to dates.
:rtype: list
"""
return self.get_new_date(datetime).to_datetime_string()
def get_new_serialized_date(self, datetime):
"""
Get the attributes that should be converted to dates.
:rtype: list
"""
return self.get_new_date(datetime).isoformat()
def set_appends(self, appends):
"""
Get the attributes that should be converted to dates.
:rtype: list
"""
self.__appends__ += appends
return self
def save_many(self, relation, relating_records):
related = getattr(self.__class__, relation)
for related_record in relating_records:
setattr(
related_record,
related.foreign_key,
self.__attributes__[related.local_key],
)
if not related_record.is_created():
related_record.create(related_record.all_attributes())
else:
related_record.save()
def related(self, relation):
related = getattr(self.__class__, relation)
return related.where(related.foreign_key, self.get_primary_key_value())
def get_related(self, relation):
related = getattr(self.__class__, relation)
return related
def attach(self, relation, related_record):
related = getattr(self.__class__, relation)
setattr(
related_record, related.foreign_key, self.__attributes__[related.local_key]
)
if not related_record.is_created():
related_record.create(related_record.all_attributes())
else:
related_record.save()
@classmethod
def on(cls, connection):
cls.__connection__ = connection
return cls