kytos-ng /
topology
| Conditions | 13 |
| Total Lines | 59 |
| Code Lines | 46 |
| Lines | 0 |
| Ratio | 0 % |
| Changes | 0 | ||
Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.
For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.
Commonly applied refactorings include:
If many parameters/temporary variables are present:
Complex classes like 001_special_vlan_allocation.update_database() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
| 1 | import datetime |
||
| 35 | def update_database(mongo: Mongo): |
||
| 36 | db = mongo.client[mongo.db_name] |
||
| 37 | intfs_documents = db.interface_details.find() |
||
| 38 | evc_documents = db.evcs.find({"archived": False}) |
||
| 39 | |||
| 40 | tag_by_intf = defaultdict(set) |
||
| 41 | evc_intf = defaultdict(str) |
||
| 42 | |||
| 43 | for evc in evc_documents: |
||
| 44 | tag_a = evc["uni_a"].get("tag") |
||
| 45 | if tag_a and isinstance(tag_a["value"], str): |
||
| 46 | intf_id = evc["uni_a"]["interface_id"] |
||
| 47 | if tag_a["value"] in tag_by_intf[intf_id]: |
||
| 48 | print(f"Error: Detected duplicated vlan '{tag_a['value']}' TAG" |
||
| 49 | f" in EVCs {evc['id']} and {evc_intf[intf_id+tag_a['value']]}" |
||
| 50 | f" in interface {intf_id}") |
||
| 51 | sys.exit(1) |
||
| 52 | tag_by_intf[intf_id].add(tag_a["value"]) |
||
| 53 | evc_intf[intf_id+tag_a["value"]] = evc["id"] |
||
| 54 | tag_z = evc["uni_z"].get("tag") |
||
| 55 | |||
| 56 | if tag_z and isinstance(tag_z["value"], str): |
||
| 57 | intf_id = evc["uni_z"]["interface_id"] |
||
| 58 | if tag_z["value"] in tag_by_intf[intf_id]: |
||
| 59 | print(f"Error: Detected duplicated vlan '{tag_z['value']}' TAG" |
||
| 60 | f" in EVCs {evc['id']} and {evc_intf[intf_id+tag_z['value']]}" |
||
| 61 | f" in interface {intf_id}") |
||
| 62 | sys.exit(1) |
||
| 63 | tag_by_intf[intf_id].add(tag_z["value"]) |
||
| 64 | evc_intf[intf_id+tag_z["value"]] = evc["id"] |
||
| 65 | |||
| 66 | default_special_vlans = {"untagged", "any"} |
||
| 67 | intf_count = 0 |
||
| 68 | message_intfs = "" |
||
| 69 | for intf in intfs_documents: |
||
| 70 | _id = intf["id"] |
||
| 71 | current_field = intf.get("special_available_tags", None) |
||
| 72 | if current_field: |
||
| 73 | current_field = set(current_field["vlan"]) |
||
| 74 | expected_field = default_special_vlans - tag_by_intf.get(_id, set()) |
||
| 75 | if current_field == expected_field and intf.get("special_tags"): |
||
| 76 | continue |
||
| 77 | db.interface_details.update_one( |
||
| 78 | {"id": _id}, |
||
| 79 | { |
||
| 80 | "$set": |
||
| 81 | { |
||
| 82 | "special_available_tags": {"vlan": list(expected_field)}, |
||
| 83 | "special_tags": {"vlan": ["untagged", "any"]} |
||
| 84 | } |
||
| 85 | } |
||
| 86 | ) |
||
| 87 | message_intfs += f"{_id}\n" |
||
| 88 | intf_count += 1 |
||
| 89 | if intf_count: |
||
| 90 | print(f"{intf_count} interface was/were updated:") |
||
| 91 | print(message_intfs) |
||
| 92 | else: |
||
| 93 | print("All interfaces are updated already.") |
||
| 94 | |||
| 110 |
Loading history...