Parse all app specifications from config. Populates self.app_specs with parsed information about each app, including the import path and worker count limits. Worker count priority: 1. Config override (e.g., "module:Class:2") - highest priority 2. Cl
(self)
| 119 | self._parse_app_specs() |
| 120 | |
| 121 | def _parse_app_specs(self): |
| 122 | """ |
| 123 | Parse all app specifications from config. |
| 124 | |
| 125 | Populates self.app_specs with parsed information about each app, |
| 126 | including the import path and worker count limits. |
| 127 | |
| 128 | Worker count priority: |
| 129 | 1. Config override (e.g., "module:Class:2") - highest priority |
| 130 | 2. Class attribute (e.g., workers = 2 on the class) |
| 131 | 3. None (all workers) - default |
| 132 | """ |
| 133 | for spec in self.cfg.dirty_apps: |
| 134 | import_path, worker_count = parse_dirty_app_spec(spec) |
| 135 | |
| 136 | # If no config override, check class attribute |
| 137 | if worker_count is None: |
| 138 | try: |
| 139 | worker_count = get_app_workers_attribute(import_path) |
| 140 | except Exception as e: |
| 141 | # Log but don't fail - we'll discover the error when loading |
| 142 | self.log.warning( |
| 143 | "Could not read workers attribute from %s: %s", |
| 144 | import_path, e |
| 145 | ) |
| 146 | |
| 147 | self.app_specs[import_path] = { |
| 148 | 'import_path': import_path, |
| 149 | 'worker_count': worker_count, |
| 150 | 'original_spec': spec, |
| 151 | } |
| 152 | # Initialize the app_worker_map for this app |
| 153 | self.app_worker_map[import_path] = set() |
| 154 | |
| 155 | def _get_minimum_workers(self): |
| 156 | """ |
no test coverage detected