<s> from __future__ import absolute_import <EOL> import os <EOL> import imp <EOL> import importlib <EOL> from datetime import datetime <EOL> from warnings import warn <EOL> from celery import signals <EOL> from celery . datastructures import DictAttribute <EOL> from celery . loaders . base import BaseLoader <EOL> import django <EOL> from django import db <EOL> from django . conf import settings <EOL> from django . core import cache <EOL> from django . core . mail import mail_admins <EOL> from . utils import DATABASE_ERRORS , now <EOL> _RACE_PROTECTION = False <EOL> NO_TZ = django . VERSION < ( <NUM_LIT:1> , <NUM_LIT:4> ) <EOL> def _maybe_close_fd ( fh ) : <EOL> try : <EOL> os . close ( fh . fileno ( ) ) <EOL> except ( AttributeError , OSError , TypeError ) : <EOL> pass <EOL> class DjangoLoader ( BaseLoader ) : <EOL> """<STR_LIT>""" <EOL> _db_reuse = <NUM_LIT:0> <EOL> override_backends = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( DjangoLoader , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _install_signal_handlers ( ) <EOL> def _install_signal_handlers ( self ) : <EOL> signals . beat_embedded_init . connect ( self . close_database ) <EOL> signals . worker_ready . connect ( self . warn_if_debug ) <EOL> def now ( self , utc = False ) : <EOL> return datetime . utcnow ( ) if utc else now ( ) <EOL> def read_configuration ( self ) : <EOL> """<STR_LIT>""" <EOL> self . configured = True <EOL> backend = ( getattr ( settings , '<STR_LIT>' , None ) or <EOL> getattr ( settings , '<STR_LIT>' , None ) ) <EOL> if not backend : <EOL> settings . CELERY_RESULT_BACKEND = '<STR_LIT>' <EOL> if NO_TZ : <EOL> if getattr ( settings , '<STR_LIT>' , None ) : <EOL> warn ( '<STR_LIT>' ) <EOL> settings . CELERY_ENABLE_UTC = False <EOL> return DictAttribute ( settings ) <EOL> def _close_database ( self ) : <EOL> try : <EOL> funs = [ conn . close for conn in db . connections ] <EOL> except AttributeError : <EOL> if hasattr ( db , '<STR_LIT>' ) : <EOL> funs = [ db . close_old_connections ] <EOL> else : <EOL> funs = [ db . close_connection ] <EOL> for close in funs : <EOL> try : <EOL> close ( ) <EOL> except DATABASE_ERRORS as exc : <EOL> str_exc = str ( exc ) <EOL> if '<STR_LIT>' not in str_exc and '<STR_LIT>' not in str_exc : <EOL> raise <EOL> def close_database ( self , ** kwargs ) : <EOL> db_reuse_max = self . conf . get ( '<STR_LIT>' , None ) <EOL> if not db_reuse_max : <EOL> return self . _close_database ( ) <EOL> if self . _db_reuse >= db_reuse_max * <NUM_LIT:2> : <EOL> self . _db_reuse = <NUM_LIT:0> <EOL> self . _close_database ( ) <EOL> self . _db_reuse += <NUM_LIT:1> <EOL> def close_cache ( self ) : <EOL> try : <EOL> cache . cache . close ( ) <EOL> except ( TypeError , AttributeError ) : <EOL> pass <EOL> def on_process_cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> self . close_database ( ) <EOL> self . close_cache ( ) <EOL> def on_task_init ( self , task_id , task ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> is_eager = task . request . is_eager <EOL> except AttributeError : <EOL> is_eager = False <EOL> if not is_eager : <EOL> self . close_database ( ) <EOL> def on_worker_init ( self ) : <EOL> """<STR_LIT>""" <EOL> self . import_default_modules ( ) <EOL> self . close_database ( ) <EOL> self . close_cache ( ) <EOL> def warn_if_debug ( self , ** kwargs ) : <EOL> if settings . DEBUG : <EOL> warn ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def import_default_modules ( self ) : <EOL> super ( DjangoLoader , self ) . import_default_modules ( ) <EOL> self . autodiscover ( ) <EOL> def autodiscover ( self ) : <EOL> self . task_modules . update ( mod . __name__ for mod in autodiscover ( ) or ( ) ) <EOL> def on_worker_process_init ( self ) : <EOL> try : <EOL> for c in db . connections . all ( ) : <EOL> if c and c . connection : <EOL> _maybe_close_fd ( c . connection ) <EOL> except AttributeError : <EOL> if db . connection and db . connection . connection : <EOL> _maybe_close_fd ( db . connection . connection ) <EOL> self . _close_database ( ) <EOL> self . close_cache ( ) <EOL> def mail_admins ( self , subject , body , fail_silently = False , ** kwargs ) : <EOL> return mail_admins ( subject , body , fail_silently = fail_silently ) <EOL> def autodiscover ( ) : <EOL> """<STR_LIT>""" <EOL> global _RACE_PROTECTION <EOL> if _RACE_PROTECTION : <EOL> return <EOL> _RACE_PROTECTION = True <EOL> try : <EOL> return filter ( None , [ find_related_module ( app , '<STR_LIT>' ) <EOL> for app in settings . INSTALLED_APPS ] ) <EOL> finally : <EOL> _RACE_PROTECTION = False <EOL> def find_related_module ( app , related_name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> app_path = importlib . import_module ( app ) . __path__ <EOL> except ImportError as exc : <EOL> warn ( '<STR_LIT>' % ( <EOL> app , related_name , exc , <EOL> ) ) <EOL> return <EOL> except AttributeError : <EOL> return <EOL> try : <EOL> imp . find_module ( related_name , app_path ) <EOL> except ImportError : <EOL> return <EOL> return importlib . import_module ( '<STR_LIT>' . format ( app , related_name ) ) </s>
