Apply ruff format auto-fixes

src/nsls2api/api/models/facility_model.py

@@ -18,10 +18,11 @@ class FacilityCurrentOperatingCycleResponseModel(pydantic.BaseModel):
     facility: str
     cycle: str
 
+
 class FacilityCycleDetailsResponseModel(pydantic.BaseModel):
     facility: str
     cycle: str
     start_date: datetime | None = None
     end_date: datetime | None = None
     is_current_operating_cycle: bool
-    accepting_proposals: bool | None = None
+    accepting_proposals: bool | None = None

src/nsls2api/api/models/person_model.py

@@ -104,13 +104,15 @@ class UnixInfo(pydantic.BaseModel):
     homeDirectory: Optional[str] = None
     loginShell: Optional[str] = None
 
+
 class IdentityInfo(pydantic.BaseModel):
     displayName: Optional[str] = None
     email: Optional[str] = None
     department: Optional[str] = None
     manager: Optional[str] = None
     unix: Optional[UnixInfo] = None
 
+
 class AccountInfo(pydantic.BaseModel):
     accountExpires: Optional[str] = None
     badPasswordTime: Optional[str] = None
@@ -126,6 +128,7 @@ class AccountInfo(pydantic.BaseModel):
     uSNCreated: Optional[int] = None
     uSNChanged: Optional[int] = None
 
+
 class DirectoryInfo(pydantic.BaseModel):
     objectGUID: Optional[str] = None
     objectSid: Optional[str] = None
@@ -134,6 +137,7 @@ class DirectoryInfo(pydantic.BaseModel):
     whenCreated: Optional[str] = None
     whenChanged: Optional[str] = None
 
+
 class AttributesInfo(pydantic.BaseModel):
     sn: Optional[str] = None
     givenName: Optional[str] = None
@@ -145,13 +149,15 @@ class AttributesInfo(pydantic.BaseModel):
     instanceType: Optional[str] = None
     objectClass: List[str] = pydantic.Field(default_factory=list)
 
+
 class LDAPUserResponse(pydantic.BaseModel):
     """Complete LDAP user data from direct LDAP query"""
+
     dn: Optional[str] = None
     status: str = "Read"
     readTime: Optional[str] = None
     identity: Optional[IdentityInfo] = None
     account: Optional[AccountInfo] = None
     directory: Optional[DirectoryInfo] = None
     groups: List[str] = pydantic.Field(default_factory=list)
-    attributes: Optional[AttributesInfo] = None
+    attributes: Optional[AttributesInfo] = None

src/nsls2api/api/models/proposal_model.py

@@ -159,8 +159,9 @@ class ProposalIdDataSession(pydantic.BaseModel):
     proposal_id: str
     data_session: str | None = None
 
+
 class ProposalIdDataSessionList(pydantic.BaseModel):
     proposals: list[ProposalIdDataSession]
     count: int
     page_size: int
-    page: int
+    page: int

src/nsls2api/api/v1/facility_api.py

@@ -3,16 +3,20 @@
 
 from nsls2api.api.models.facility_model import (
     FacilityCurrentOperatingCycleResponseModel,
-    FacilityCycleDetailsResponseModel, FacilityCyclesResponseModel,
-    FacilityName)
+    FacilityCycleDetailsResponseModel,
+    FacilityCyclesResponseModel,
+    FacilityName,
+)
 from nsls2api.api.models.proposal_model import CycleProposalList
 from nsls2api.infrastructure.logging import logger
 from nsls2api.infrastructure.security import validate_admin_role
 from nsls2api.services import facility_service, proposal_service
-from nsls2api.services.facility_service import (CycleNotFoundError,
-                                                CycleOperationError,
-                                                CycleUpdateError,
-                                                CycleVerificationError)
+from nsls2api.services.facility_service import (
+    CycleNotFoundError,
+    CycleOperationError,
+    CycleUpdateError,
+    CycleVerificationError,
+)
 
 router = fastapi.APIRouter()
 

src/nsls2api/api/v1/proposal_api.py

@@ -15,7 +15,7 @@
     RecentProposalsList,
     SingleProposal,
     UsernamesList,
-    ProposalIdDataSessionList
+    ProposalIdDataSessionList,
 )
 from nsls2api.infrastructure.logging import logger
 from nsls2api.infrastructure.security import get_current_user, validate_admin_role
@@ -105,13 +105,14 @@ async def get_proposals(
     cycle: Annotated[list[str], Query()] = [],
     facility: Annotated[list[FacilityName], Query()] = [FacilityName.nsls2],
     username: str | None = Query(None, description="Filter proposals by username"),
-    saf_status: list[str] | None = Query(default=None, description="Filter proposals and SAFs by SAF status"),
+    saf_status: list[str] | None = Query(
+        default=None, description="Filter proposals and SAFs by SAF status"
+    ),
     page_size: int = Query(10, ge=1, le=200),
     page: int = Query(1, ge=1),
     include_directories: bool = False,
 ):
 
-
     proposal_list = await proposal_service.fetch_proposals(
         proposal_id=proposal_id,
         beamline=beamline,
@@ -123,7 +124,6 @@ async def get_proposals(
         page=page,
         include_directories=include_directories,
     )
-
 
     response_model = {
         "proposals": proposal_list,
@@ -135,7 +135,6 @@ async def get_proposals(
     return response_model
 
 
-
 @router.get(
     "/proposals/data-sessions",
     response_model=ProposalIdDataSessionList,
@@ -156,7 +155,7 @@ async def get_proposals_data_sessions(
         cycle=cycle,
         facility=facility,
         page_size=page_size,
-        page=page
+        page=page,
     )
 
     response_model = {

src/nsls2api/api/v1/user_api.py

@@ -73,25 +73,26 @@ async def get_person_by_department(department_code: str = "PS"):
 
 
 # TODO: Add back into schema if we decide to use this endpoint.
-@router.get("/person/me",include_in_schema=True)
+@router.get("/person/me", include_in_schema=True)
 async def get_myself(upn: str = Header(...)):
-    #upn: User principal name
+    # upn: User principal name
     if not upn:
-        raise HTTPException(status_code=400, detail = "upn not found")
+        raise HTTPException(status_code=400, detail="upn not found")
     settings = get_settings()
-    ldap_info = await asyncio.to_thread(get_user_info,
+    ldap_info = await asyncio.to_thread(
+        get_user_info,
         upn,
         settings.ldap_server,
         settings.ldap_base_dn,
         settings.ldap_bind_user,
-        settings.ldap_bind_password
+        settings.ldap_bind_password,
     )
     if not ldap_info:
         raise HTTPException(status_code=404, detail="User not found in LDAP")
-    
+
     shaped_info = shape_ldap_response(ldap_info)
     return LDAPUserResponse(**shaped_info)
-    
+
 
 @router.get("/data-session/{username}", response_model=DataSessionAccess, tags=["data"])
 @router.get(

src/nsls2api/infrastructure/config.py

@@ -71,8 +71,10 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
-    #Whoami LDAP settings
-    ldap_server: str = Field(default="ldaps://ldapproxy.nsls2.bnl.gov", alias="LDAP_SERVER")
+    # Whoami LDAP settings
+    ldap_server: str = Field(
+        default="ldaps://ldapproxy.nsls2.bnl.gov", alias="LDAP_SERVER"
+    )
     ldap_base_dn: str = Field(default="dc=bnl,dc=gov", alias="LDAP_BASE_DN")
     ldap_bind_user: str = Field(default="", alias="LDAP_BIND_USER")
     ldap_bind_password: str = Field(default="", alias="LDAP_BIND_PASSWORD")

src/nsls2api/services/beamline_service.py

@@ -77,7 +77,9 @@ async def all_services(name: str) -> Optional[ServicesOnly]:
 
 
 async def detectors(name: str) -> list[Detector]:
-    beamline_detectors = await Beamline.find_one(Beamline.name == name.upper()).project(DetectorView)
+    beamline_detectors = await Beamline.find_one(Beamline.name == name.upper()).project(
+        DetectorView
+    )
     if beamline_detectors is None:
         raise LookupError(f"Beamline '{name.upper()}' does not exist.")
     return beamline_detectors.detectors

src/nsls2api/services/ldap_service.py

@@ -7,30 +7,33 @@
 
 
 def to_hex(val):
-    
+
     if isinstance(val, bytes):
         return binascii.hexlify(val).decode()
     return None
 
+
 def get_user_info(upn, ldap_server, ldap_base_dn, ldap_bind_user, bind_password):
-    conn = None 
+    conn = None
     try:
         server = Server(ldap_server)
-        conn = Connection(server, user=ldap_bind_user, password=bind_password, auto_bind=True)
+        conn = Connection(
+            server, user=ldap_bind_user, password=bind_password, auto_bind=True
+        )
         search_filter = f"(&(objectclass=person)(userPrincipalName={upn}))"
-        conn.search(ldap_base_dn, search_filter, attributes=['sAMAccountName'])
+        conn.search(ldap_base_dn, search_filter, attributes=["sAMAccountName"])
 
         if not conn.entries:
             logger.warning("No entries found for the given UPN.")
             return None
 
         entry = conn.entries[0]
-        username = entry.sAMAccountName.value if 'sAMAccountName' in entry else None
+        username = entry.sAMAccountName.value if "sAMAccountName" in entry else None
         if username is None:
             return None
 
         search_filter = f"(&(objectclass=posixaccount)(sAMAccountName={username}))"
-        conn.search(ldap_base_dn, search_filter, attributes=['*'])
+        conn.search(ldap_base_dn, search_filter, attributes=["*"])
 
         if not conn.entries:
             logger.warning("no posix entries found for the given username.")
@@ -52,44 +55,60 @@ def get_user_info(upn, ldap_server, ldap_base_dn, ldap_bind_user, bind_password)
         if conn is not None:
             conn.unbind()
 
+
 def filetime_to_str(filetime):
     try:
-        if filetime is None or int(filetime) == 0 or int(filetime) == 9223372036854775807:
+        if (
+            filetime is None
+            or int(filetime) == 0
+            or int(filetime) == 9223372036854775807
+        ):
             return "Never"
         dt = datetime(1601, 1, 1) + timedelta(microseconds=int(filetime) // 10)
         return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
     except Exception:
         return str(filetime)
 
+
 def generalized_time_to_str(gt):
     try:
-        if not gt: return ""
+        if not gt:
+            return ""
         dt = datetime.strptime(gt.split(".")[0], "%Y%m%d%H%M%S")
         return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
     except Exception:
         return str(gt)
 
+
 def decode_uac(uac):
     flags = []
     try:
         val = int(uac)
-        if val & 0x0001: flags.append("SCRIPT")
-        if val & 0x0002: flags.append("ACCOUNTDISABLE")
-        if val & 0x0008: flags.append("HOMEDIR_REQUIRED")
-        if val & 0x0200: flags.append("NORMAL_ACCOUNT")
-        if val & 0x1000: flags.append("PASSWORD_EXPIRED")
+        if val & 0x0001:
+            flags.append("SCRIPT")
+        if val & 0x0002:
+            flags.append("ACCOUNTDISABLE")
+        if val & 0x0008:
+            flags.append("HOMEDIR_REQUIRED")
+        if val & 0x0200:
+            flags.append("NORMAL_ACCOUNT")
+        if val & 0x1000:
+            flags.append("PASSWORD_EXPIRED")
     except Exception:
         return []
     return flags or ["NORMAL_ACCOUNT"]
 
+
 def shape_ldap_response(user_info, dn=None, status="Read", read_time=None):
     def clean_groups(groups_val):
         if not groups_val:
             return []
         if isinstance(groups_val, list):
             return groups_val
         elif isinstance(groups_val, str):
-            return [g.strip() for g in groups_val.replace("\n", ",").split(",") if g.strip()]
+            return [
+                g.strip() for g in groups_val.replace("\n", ",").split(",") if g.strip()
+            ]
         return []
 
     return {
@@ -106,8 +125,8 @@ def clean_groups(groups_val):
                 "uidNumber": user_info.get("uidNumber"),
                 "gidNumber": user_info.get("gidNumber"),
                 "homeDirectory": user_info.get("homeDirectory"),
-                "loginShell": user_info.get("loginShell")
-            }
+                "loginShell": user_info.get("loginShell"),
+            },
         },
         "account": {
             "accountExpires": filetime_to_str(user_info.get("accountExpires")),
@@ -142,6 +161,8 @@ def clean_groups(groups_val):
             "codePage": user_info.get("codePage"),
             "countryCode": user_info.get("countryCode"),
             "instanceType": user_info.get("instanceType"),
-            "objectClass": [s.strip() for s in user_info.get("objectClass", "").split() if s.strip()]
-        }
-    }
+            "objectClass": [
+                s.strip() for s in user_info.get("objectClass", "").split() if s.strip()
+            ],
+        },
+    }